This class is part of a utility that reads lines from a set of nonblocking file descriptors, blocking only when there are no complete lines to emit.
class NonblockingLineBuffer: def __init__(self, fd, encoding): self.fd = fd self.enc = encoding self.buf = bytearray() def absorb(self): while True: try: block = os.read(self.fd, 8192) except BlockingIOError: return if block: self.buf.extend(block) else: self.is_open = False # We don't close the file here because caller # needs to remove the fd from the poll set first. return def emit(self): def emit1(chunk): self.emitted_this_cycle = True return (self.fd, chunk.decode(self.enc).rstrip()) buf = self.buf self.emitted_this_cycle = False while buf: r = buf.find(b'\r') n = buf.find(b'\n') if r == -1 and n == -1: if not self.is_open: yield emit1(buf) buf.clear() elif r == -1 or r > n: yield emit1(buf[:n]) buf = buf[(n+1):] elif n == -1 or n > r: yield emit1(buf[:r]) if n == r+1: buf = buf[(r+2):] else: buf = buf[(r+1):] self.buf = buf if not self.is_open: self.emitted_this_cycle = True yield (self.fd, None)
This question is specifically about
emit, which is complicated, confusing, and might not be as efficient as it could be. Please suggest ways to make it less complicated and/or confusing, and more efficient.
(I know it could be much simpler if it didn't need to reimplement universal newline handling, but that is unfortunately a requirement from the larger context.)
(If there's something in the standard library that does some or all of the larger task, that would also be a welcome answer.)