Однако я не знаю, когда у reader есть данные для чтения. Что я могу с этим сделать?
Очевидный способ узнать, когда поток считывателя имеет данные для чтения, - это await it:
data = await reader.read(1024)
Это либо вернет данные сразу, либо приостановит текущую сопрограмму, позволяя другим сопрограммам добиться прогресса, и возобновит ее только тогда, когда у читателя есть какие-то данные для чтения. Вместо того, чтобы хранить устройство чтения / записи для будущего общения, вы можете написать сопрограмму, которая выполняет сообщение, и сохранить задачу 1015 *, которая ее управляет:
async def communicate():
reader, writer = await asyncio.open_connection(addr)
# an echo server
while True:
line = await reader.readline()
if not line:
break
writer.write(line)
await writer.drain() # backpressure, see https://tinyurl.com./hqylfay
task = loop.create_task(communicate())
# the task can itself be awaited, canceled, etc.
Идея asyncio stream API состоит в том, чтобы написать такой последовательный код, оставляя asyncio обрабатывать опросы файловых дескрипторов и планировать задачи. Вы можете использовать комбинаторы, такие как asyncio.gather и asyncio.wait для параллельного запуска тысяч таких легких сопрограмм.
Есть ли способ сделать обработчик, когда ридер готов?
Если вам нужен API на основе обратного вызова, вам, вероятно, следует вместо этого использовать транспорты и протоколы более низкого уровня. Однако, если вы уже работаете с потоками, но все еще время от времени нуждаетесь в обычном обратном вызове, вы можете получить его, получив Future:
future = asyncio.ensure_future(reader.read(1024))
future.add_done_callback(your_callback)
Future имеет роль, эквивалентную обработчику сопрограмм. Как только read больше не будет блокироваться, цикл обработки вызовов будет вызываться циклом обработки событий с единственным аргументом - будущее. Будущее закончится, и его метод result() может быть использован для извлечения полученных данных или исключения.
(Вышеприведенное относится к любому сопрограммному объекту или объекту, совместимому с будущим в asyncio, а не только к StreamReader методам.)