Async variant of :meth:`flush_sync`.
(self, send: typing.Callable[[str], typing.Awaitable[object]])
| 57 | raise |
| 58 | |
| 59 | async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object]]) -> None: |
| 60 | """Async variant of :meth:`flush_sync`.""" |
| 61 | with self._lock: |
| 62 | pending = list(self._queue) |
| 63 | self._queue.clear() |
| 64 | self._bytes = 0 |
| 65 | |
| 66 | for i, (data, _byte_length) in enumerate(pending): |
| 67 | try: |
| 68 | await send(data) |
| 69 | except Exception: |
| 70 | with self._lock: |
| 71 | remaining = pending[i:] |
| 72 | self._queue = remaining + self._queue |
| 73 | self._bytes = sum(bl for _, bl in self._queue) |
| 74 | raise |
| 75 | |
| 76 | def drain(self) -> list[str]: |
| 77 | """Remove and return all queued messages.""" |