(self)
| 469 | self.sending_queue = Queue() if fuse_message else None |
| 470 | |
| 471 | def setup_sender(self): |
| 472 | if not self.fuse_message or self._send_thread is not None: |
| 473 | return |
| 474 | |
| 475 | def send_task(): |
| 476 | while True: |
| 477 | qsize = self.sending_queue.qsize() |
| 478 | if qsize > 0: |
| 479 | qsize = min(self.fuse_size, qsize) |
| 480 | self._obj_counter += qsize |
| 481 | message = [ |
| 482 | self.sending_queue.get_nowait() for _ in range(qsize) |
| 483 | ] |
| 484 | self.queue.put(message) |
| 485 | self._message_counter += 1 |
| 486 | else: |
| 487 | time.sleep(0.001) |
| 488 | |
| 489 | self._send_thread = ManagedThread(send_task, |
| 490 | name="fused_send_thread", |
| 491 | error_queue=self.error_queue) |
| 492 | self._send_thread.start() |
| 493 | |
| 494 | def put(self, obj: Any): |
| 495 | self.setup_sender() |
no test coverage detected