code/solutions/framing_lab.py

"""C12.05: bounded length-prefixed framing over arbitrary byte chunks."""
import struct
MAX_PAYLOAD = 4096

def frame(payload: bytes) -> bytes:
    if len(payload) > MAX_PAYLOAD:
        raise ValueError("frame too large")
    return struct.pack(">I", len(payload)) + payload

class Decoder:
    def __init__(self):
        self._buffer = bytearray()
        self._needed: int | None = None
        self._failed = False

    def feed(self, chunk: bytes) -> list[bytes]:
        if self._failed:
            raise ValueError("decoder is failed")
        out = []
        # Consume incrementally: even a large caller chunk does not cause
        # an unbounded internal buffering allocation.
        for byte in chunk:
            self._buffer.append(byte)
            if self._needed is None and len(self._buffer) == 4:
                self._needed = struct.unpack(">I", self._buffer)[0]
                self._buffer.clear()
                if self._needed > MAX_PAYLOAD:
                    self._failed = True
                    raise ValueError("advertised frame too large")
            if self._needed is not None and len(self._buffer) == self._needed:
                out.append(bytes(self._buffer))
                self._buffer.clear()
                self._needed = None
        return out

    def finish(self) -> None:
        if self._failed or self._buffer or self._needed is not None:
            raise ValueError("truncated or invalid stream")

if __name__ == "__main__":
    d = Decoder()
    wire = frame(b"hello") + frame(b"")
    print(d.feed(wire[:3]) + d.feed(wire[3:]))
    d.finish()