janus_thread_safe_queue_sync_async_communication.py

1import asyncio
2import janus
3
4async def main():
5    # A janus queue must be created within an asyncio event loop.
6    queue = janus.Queue()
7
8    def sync_worker():
9        # This function runs in a separate thread and uses the synchronous interface.
10        for i in range(5):
11            queue.sync_q.put(i)
12        # Signal the end of work
13        queue.sync_q.put(None)
14
15    async def async_worker():
16        # This function runs in the main asyncio loop and uses the asynchronous interface.
17        while True:
18            val = await queue.async_q.get()
19            if val is None:
20                queue.async_q.task_done()
21                break
22            print(f"Async worker received: {val}")
23            queue.async_q.task_done()
24
25    # Get the current event loop to run the synchronous executor
26    loop = asyncio.get_running_loop()
27    
28    # Run the synchronous worker in a thread pool executor
29    sync_fut = loop.run_in_executor(None, sync_worker)
30    
31    # Run the asynchronous worker
32    await async_worker()
33    
34    # Wait for the synchronous thread to finish
35    await sync_fut
36    
37    # Cleanly close the queue
38    queue.close()
39    await queue.wait_closed()
40
41if __name__ == "__main__":
42    asyncio.run(main())