Send every queued message via *send*. If *send* raises, the failing message and all subsequent messages are re-queued and the error is re-raised.
(self, send: typing.Callable[[str], object])
| 36 | self._bytes += byte_length |
| 37 | |
| 38 | def flush_sync(self, send: typing.Callable[[str], object]) -> None: |
| 39 | """Send every queued message via *send*. |
| 40 | |
| 41 | If *send* raises, the failing message and all subsequent messages |
| 42 | are re-queued and the error is re-raised. |
| 43 | """ |
| 44 | with self._lock: |
| 45 | pending = list(self._queue) |
| 46 | self._queue.clear() |
| 47 | self._bytes = 0 |
| 48 | |
| 49 | for i, (data, _byte_length) in enumerate(pending): |
| 50 | try: |
| 51 | send(data) |
| 52 | except Exception: |
| 53 | with self._lock: |
| 54 | remaining = pending[i:] |
| 55 | self._queue = remaining + self._queue |
| 56 | self._bytes = sum(bl for _, bl in self._queue) |
| 57 | raise |
| 58 | |
| 59 | async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object]]) -> None: |
| 60 | """Async variant of :meth:`flush_sync`.""" |