Skip to content

Instantly share code, notes, and snippets.

@bwangelme
Forked from nhoad/gist:8966377
Created July 9, 2017 15:48
Show Gist options
  • Save bwangelme/451431531054996c26fe93365246d2aa to your computer and use it in GitHub Desktop.
Save bwangelme/451431531054996c26fe93365246d2aa to your computer and use it in GitHub Desktop.
Async stdio with asyncio
import os
import asyncio
import sys
from asyncio.streams import StreamWriter, FlowControlMixin
reader, writer = None, None
@asyncio.coroutine
def stdio(loop=None):
if loop is None:
loop = asyncio.get_event_loop()
reader = asyncio.StreamReader()
reader_protocol = asyncio.StreamReaderProtocol(reader)
writer_transport, writer_protocol = yield from loop.connect_write_pipe(FlowControlMixin, os.fdopen(0, 'wb'))
writer = StreamWriter(writer_transport, writer_protocol, None, loop)
yield from loop.connect_read_pipe(lambda: reader_protocol, sys.stdin)
return reader, writer
@asyncio.coroutine
def async_input(message):
if isinstance(message, str):
message = message.encode('utf8')
global reader, writer
if (reader, writer) == (None, None):
reader, writer = yield from stdio()
writer.write(message)
yield from writer.drain()
line = yield from reader.readline()
return line.decode('utf8').replace('\r', '').replace('\n', '')
@asyncio.coroutine
def main():
name = yield from async_input("What's your name? ")
print("Hello, {}!".format(name))
asyncio.get_event_loop().run_until_complete(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment