| """Bridge a blocking token generator onto an asyncio queue.""" |
|
|
| import threading |
| from asyncio import AbstractEventLoop, Queue |
| from collections.abc import Iterator |
|
|
|
|
| def pump(chunks: Iterator[str], queue: Queue, loop: AbstractEventLoop) -> threading.Thread: |
| """Forward every item of `chunks` to `queue` from a daemon thread. |
| |
| The stream, queue and loop are arguments rather than closed-over locals on |
| purpose. A caller that abandons a pump — the chat fallback loop does, on its |
| TTFT timeout — leaves behind a thread writing into a queue nobody reads, |
| instead of one that writes into whatever the next attempt rebound those |
| names to. |
| |
| Exceptions from the stream are forwarded as values; `None` ends the stream. |
| |
| Returns the running thread. |
| """ |
|
|
| def run() -> None: |
| try: |
| for chunk in chunks: |
| loop.call_soon_threadsafe(queue.put_nowait, chunk) |
| except Exception as error: |
| loop.call_soon_threadsafe(queue.put_nowait, error) |
| finally: |
| loop.call_soon_threadsafe(queue.put_nowait, None) |
|
|
| thread = threading.Thread(target=run, daemon=True) |
| thread.start() |
| return thread |
|
|
