从流中产生的正确方法是什么？

Question

我有一个Connection 对象，用于包含asyncio 连接的读写流：

class Connection(object):

    def __init__(self, stream_in, stream_out):
        object.__init__(self)

        self.__in = stream_in
        self.__out = stream_out

    def read(self, n_bytes : int = -1):
        return self.__in.read(n_bytes)

    def write(self, bytes_ : bytes):
        self.__out.write(bytes_)
        yield from self.__out.drain()

在服务器端，connected 每次客户端连接时都会创建一个Connection 对象，然后读取 4 个字节。

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

在客户端，4个字节被写入。

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

这几乎可以按预期工作，但我必须在每次 read 和 write 通话时拨打 yield from。我从Connection 内部尝试过yield froming：

def read(self, n_bytes : int = -1):
    data = yield from self.__in.read(n_bytes)
    return data

但我得到的不是数据，而是类似的输出

<generator object StreamReader.read at 0x1109983b8>

如果我从多个地方调用read 和write，我宁愿不要每次都重复yield froms；而是将它们留在Connection 内。我的最终目标是将我的new_conection 函数缩减为：

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

Answer 1

因为 StreamReader.read is a coroutine，调用它的唯一选择是 a) 将其包装在 Task 或 Future 中并通过事件循环运行它，b) await从使用 async def 定义的协程中运行它, 或 c) 在协程中使用 yield from 和它，该协程定义为用 @asyncio.coroutine 装饰的函数。

由于Connection.read 是从事件循环中调用的（通过协程new_connection），因此您不能重用该事件循环来运行Task 或Future for StreamReader.read：event loops can't be started while they're already running。您要么必须stop the event loop（灾难性的，可能无法正确执行）或create a new event loop（混乱并违背使用协程的目的）。这些都不是可取的，所以Connection.read 需要是协程或async 函数。

其他两个选项（async def 协程中的await 或@asyncio.coroutine 装饰函数中的yield from）大部分是等效的。唯一的区别是async def and await were added in Python 3.5，所以对于3.4，使用yield from和@asyncio.coroutine是唯一的选择（协程和asyncio在3.4之前不存在，所以其他版本无关紧要）。就个人而言，我更喜欢使用async def 和await，因为使用async def 定义协程比使用装饰器更清晰。

简而言之：让Connection.read 和new_connection 成为协程（使用装饰器或async 关键字），并在调用其他协程时使用await（或yield from）（await conn.read(4) in @ 987654360@ 和await self.__in.read(n_bytes) 在Connection.read)。

Answer 2

我发现第 620 行的 StreamReader source code 的一部分实际上是该函数用法的完美示例。

在我之前的回答中，我忽略了这样一个事实，即self.__in.read(n_bytes) 不仅是一个协程（考虑到它来自asyncio 模块 XD，我应该知道它），而且它会产生在线结果。所以它实际上是一个生成器，你需要从中屈服。

从源代码中借用这个循环，你的读取函数应该是这样的：

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

因为self.__in.read(n_bytes) 是一个生成器，所以您必须继续从它产生，直到它产生一个空结果以表示读取结束。现在您的读取函数应该返回数据而不是生成器。你不必放弃这个版本的conn.read()。