How to use a read/write stream between two Python asyncio coroutines?

user_141895 0
Viewed 5697

How can I use asyncio to implement a pipe between two coroutines, one that reads from a stream and the other that writes into it?

Suppose we have this existing code, two simple scripts. One that produces to stdout:

# produce.py

import asyncio
import random
import sys

async def produce(stdout):
    for i in range(10000):
        await asyncio.sleep(random.randint(0, 3))
        print(i, file=stdout, flush=True)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(produce(sys.stdout))
    loop.close()

And the other that reads from stdin:

# consume.py

async def consume(loop, stdin):
    reader = asyncio.StreamReader(loop=loop)
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: reader_protocol, stdin)

    while True:
        line = await reader.readline()
        if not line:
            break
        print(int(line) ** 2)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(consume(loop, sys.stdin))
    loop.close()

Obviously, since our two pieces can run individually from the command-line, we could use the subprocess module with shell pipes (produce | consume).

But we would like to implement the equivalent of a Unix pipe in Python, i.e. connect the streams of those two existing functions.

Something like this won't work:

pipe = io.BytesIO()

await asyncio.gather(produce(pipe),
                     consume(loop, pipe))

If the two functions would manipulate generators, we could write something like this (python 3.6):

async def produce():
    for i in range(10000):
        await asyncio.sleep(random.randint(0, 3))
        yield str(i)


async def consume(generator):
    async for value in generator:
        print(int(value) ** 2)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(consume(produce()))
    loop.close()

Is there some parts of the asyncio API that would allow that ?

Thanks!

python-asyncio pipe python
2 Answers

The original post says "Something like this won't work." I'm not sure if this statement is indented to mean "the following code did not work" or "I do not want a solution of this style."

I will note that the following code does work:

r, w = os.pipe()
read_pipe = os.fdopen(r, 'r')
write_pipe = os.fdopen(w, 'w')

await asyncio.gather(produce(write_pipe), consume(loop, read_pipe))
user_8592770
Related