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

Question

asyncio有StreamReader.readline(),允许这样的东西:

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

(我认为async for在asyncio中没有,但那将是明显的演变)

如何实现三重奏的等效？

在三人组0.9中,我没有直接看到任何高级别的支持.我所看到的只是ReceiveStream.receive_some()返回任意大小的二进制块; 对我来说,解码并将其转换为线性似乎并非易事.我可以使用标准库函数或代码片段吗？我发现io stdlib模块看起来很有前途,但我认为没有办法提供"feed"方法.

Answer 1

你是对的,目前Trio中没有高级支持.应该有一些东西,虽然我不是100%肯定它应该是什么样子.我开了一个问题来讨论它.

与此同时,您的实施看起来很合理.

如果你想让它更加健壮,你可以(1)使用a 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

(根据您喜欢的许可,随意使用.)