What is the correct way to yield from a stream?

Question

I have a Connection type that I’m using to wrap read/write stream pairs from asyncio.

class Connection(object):

    def __init__(self, stream_in, stream_out):
        self._streams_ = (stream_in, stream_out)

    def read(self, n_bytes: int = -1):
        stream = self._streams_[0]
        return stream.read(n_bytes)

    def write(self, bytes_: bytes):
        stream = self._streams_[1]
        stream.write(bytes_)
        yield from stream.drain()

When a new client connects to the server, new_connection will create a new Connection object, and expect to receive 4 bytes.

@asyncio.coroutine
def new_connection(stream_in, stream_out):
    conn = Connection(stream_in, stream_out)
    data = yield from conn.read(4)
    print(data)

The client sends 4 bytes.

@asyncio.coroutine
def client(loop):
    ...
    conn = Connection(stream_in, stream_out)
    yield from conn.write(b'test')

This works about as I expect, but I do have to write yield from for every call to read and write. I've tried moving the yield from into Connection for this reason.

def read(self, n_bytes: int = -1):
    stream = self._streams_[0]
    data = yield from stream.read(n_bytes)
    return data

But, instead of the expected data bytes, I get a generator object.

<generator object StreamReader.read at 0x1109983b8>

So, for every call to read and write, I must be careful to have the yield from. My goal is to reduce new_connection to the following.

@asyncio.coroutine
def new_connection(stream_in, stream_out):
    conn = Connection(stream_in, stream_out)
    print(conn.read(4))

Answer 1

Because StreamReader.read is a coroutine, your only options for calling it are a) wrapping it in a Task or Future and running that via an event loop, b) awaiting it from coroutine defined with async def, or c) using yield from with it from a coroutine defined as a function decorated with @asyncio.coroutine.

Since Connection.read is called from an event loop (via the coroutine new_connection), you can't reuse that event loop to run a Task or Future for StreamReader.read: event loops can't be started while they're already running. You'd either have to stop the event loop (disastrous and probably not possible to do correctly) or create a new event loop (messy and defeating the purpose of using coroutines). Neither of those are desirable, so Connection.read needs to be a coroutine or an async function.

The other two options (await in an async def coroutine or yield from in a @asyncio.coroutine-decorated function) are mostly equivalent. The only difference is that async def and await were added in Python 3.5, so for 3.4, using yield from and @asyncio.coroutine is the only option (coroutines and asyncio didn't exist prior to 3.4, so other versions are irrelevant). Personally, I prefer using async def and await, because defining coroutines with async def is cleaner and clearer than with the decorator.

In brief: have Connection.read and new_connection be coroutines (using either the decorator or the async keyword), and use await (or yield from) when calling other coroutines (await conn.read(4) in new_connection, and await self.__in.read(n_bytes) in Connection.read).

Answer 2

I found a chunk of the StreamReader source code on line 620 is actually a perfect example of the function's usage.

In my previous answer, I overlooked the fact that self.__in.read(n_bytes) is not only a coroutine (which I should've known considering it was from the asyncio module XD) but it yields a result on line . So it is in fact a generator, and you will need to yield from it.

Borrowing this loop from the source code, your read function should look something like this:

def read(self, n_bytes : int = -1):
    data = bytearray() #or whatever object you are looking for
    while 1:
        block = yield from self.__in.read(n_bytes)
        if not block:
            break
        data += block
    return data

Because self.__in.read(n_bytes) is a generator, you have to continue to yield from it until it yields an empty result to signal the end of the read. Now your read function should return data rather than a generator. You won't have to yield from this version of conn.read().