3 ms·
I wonder how this fails then: import asyncio import os import signal import sys async def get_stdin_reader(): loop = asyncio.get_running_
by badmintonbaseba 2y ago
I wonder how this fails then:
import asyncio
import os
import signal
import sys
async def get_stdin_reader():
loop = asyncio.get_running_loop()
reader = asyncio.StreamReader()
protocol = asyncio.StreamReaderProtocol(reader)
await loop.connect_read_pipe(lambda: protocol, sys.stdin)
return reader
async def amain():
print(f'pid: {os.getpid()}')
sig_queue: asyncio.Queue[None] = asyncio.Queue()
def sig_handler() -> None:
sig_queue.put_nowait(None)
loop= asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGUSR1, sig_handler)
reader = await get_stdin_reader()
task1 = loop.create_task(reader.read(1))
task2 = loop.create_task(sig_queue.get())
pending = {task1, task2}
while True:
done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
if task1 in done:
task1 = loop.create_task(reader.read(1))
pending.add(task1)
print('got char on stdin')
if task2 in done:
task2 = loop.create_task(sig_queue.get())
pending.add(task2)
print('got signal')
asyncio.run(amain())