trio-line-receiver.py

# Really need to factor out a bunch of this into reusable logic
# That's what I eventually want to do with the sansio_toolbelt library
# But for now, this may be handy...

async def receive_lines(rstream, *, receive_size=16384, max_line_size=16384):
    buf = bytearray()
    searched = 0

    while True:
        found = buf.find(b"\n", searched)
        if found == -1:
            # No newline yet
            if len(buf) > max_line_size:
                raise RuntimeError("line is toooooo looooonnng")
            searched = len(buf)
            new_data = await rstream.receive_some(receive_size)
            if not new_data:
                # EOF
                if buf:
                    raise RuntimeError("final line truncated?!")
                return
            buf += new_data
        else:
            yield buf[:found]
            searched = 0
            del buf[:found + 1]
            
# Usage:
async for line in receive_lines(stream):
    ...