如何从三个 ReceiveStream 中一次读取一行？

异步有StreamReader.readline() https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamReader.readline，允许类似：

while True:
    line = await reader.readline()
    ...

（我没看到async for在 asyncio 中可用，但这将是明显的演变）

我怎样才能达到三重奏的同等效果？

我在 trio 0.9 中没有直接看到任何对此的高级支持。我所看到的只是ReceiveStream.receive_some() https://trio.readthedocs.io/en/latest/reference-io.html#trio.abc.ReceiveStream.receive_some它返回任意大小的二进制块；对我来说，解码并将其转换为逐行的东西似乎并不简单。是否有我可以使用的标准库函数或代码片段？我发现 io stdlib 模块看起来很有前途，但我没有看到任何提供“feed”方法的方法。

你是对的，目前 Trio 中没有对此提供高级支持。应该有某物，尽管我不是 100% 确定它应该是什么样子。我打开了an issue https://github.com/python-trio/trio/issues/796来讨论它。

同时，您的实现看起来很合理。

如果你想让它更加健壮，你可以（1）使用bytearray代替bytes对于你的缓冲区，要使追加和删除分摊为 O(n) 而不是 O(n^2)，(2) 对最大行长度设置限制，因此邪恶的同行不能强迫你浪费无限内存缓冲无限长行，(3) 恢复每个调用find在最后一个停止的地方，而不是每次从头开始，再次避免 O(n^2) 行为。如果您只与合理的线路长度和行为良好的同行打交道，那么这些都不是超级重要的，但也没有坏处。

这是代码的调整版本，尝试合并这三个想法：

class LineReader:
    def __init__(self, stream, max_line_length=16384):
        self.stream = stream
        self._line_generator = self.generate_lines(max_line_length)

    @staticmethod
    def generate_lines(max_line_length):
        buf = bytearray()
        find_start = 0
        while True:
            newline_idx = buf.find(b'\n', find_start)
            if newline_idx < 0:
                # no b'\n' found in buf
                if len(buf) > max_line_length:
                    raise ValueError("line too long")
                # next time, start the search where this one left off
                find_start = len(buf)
                more_data = yield
            else:
                # b'\n' found in buf so return the line and move up buf
                line = buf[:newline_idx+1]
                # Update the buffer in place, to take advantage of bytearray's
                # optimized delete-from-beginning feature.
                del buf[:newline_idx+1]
                # next time, start the search from the beginning
                find_start = 0
                more_data = yield line

            if more_data is not None:
                buf += bytes(more_data)

    async def readline(self):
        line = next(self._line_generator)
        while line is None:
            more_data = await self.stream.receive_some(1024)
            if not more_data:
                return b''  # this is the EOF indication expected by my caller
            line = self._line_generator.send(more_data)
        return line

（您可以根据您喜欢的任何许可证随意使用。）