(work_queue: queue.Queue, item: ty.Any)
| 245 | return thread |
| 246 | |
| 247 | def checked_put(work_queue: queue.Queue, item: ty.Any): |
| 248 | error = None |
| 249 | while True: |
| 250 | try: |
| 251 | work_queue.put(item, timeout=0.1) |
| 252 | return |
| 253 | except queue.Full: |
| 254 | error = check_error_queue() |
| 255 | if error is not None: |
| 256 | break |
| 257 | continue |
| 258 | raise error[1].with_traceback(error[2]) |
| 259 | |
| 260 | encode_thread = launch_thread( |
| 261 | self.image_encode_thread, |
nothing calls this directly
no outgoing calls
no test coverage detected