Handles encoding and decoding of network messages using msgpack serialization with length-prefix framing.
The protocol uses a 4-byte length-prefix framing format:
┌──────────────┬────────────────────────┐
│ 4 bytes │ N bytes │
│ length (BE) │ msgpack payload │
└──────────────┴────────────────────────┘
- Efficient: O(1) boundary detection
- Safe: No delimiter collision risk
- Standard: Used by Kafka, Redis, Protocol Buffers, etc.
Functions
encode()
def encode(data: dict) -> bytes
Dictionary containing the message data. Can include any msgpack-serializable types (dict, list, str, int, float, bool, None).
The encoded message with its 4-byte length prefix.
TypeError: If data contains non-serializable types.
Example:
from repod.protocol import encode
encoded = encode({"action": "ping", "count": 5})
len(encoded) > 4 # includes the 4-byte header
# True
decode()
def decode(data: bytes) -> dict
This expects raw msgpack data, not a full length-prefixed frame. Use read_message() for stream-based decoding.
Raw msgpack-serialized bytes.
The decoded message dictionary.
msgpack.UnpackException: If data is not valid msgpack.
Example:
import msgpack
from repod.protocol import decode
raw = msgpack.packb({"action": "hello"})
decode(raw)
# {'action': 'hello'}
read_message()
def read_message(stream: bytes) -> tuple[dict | None, int]
Byte buffer that may contain partial or complete messages.
A (message, bytes_consumed) tuple. If the stream does not yet contain a full message, returns (None, 0).
from repod.protocol import encode, read_message
frame = encode({"action": "test"})
msg, consumed = read_message(frame)
msg
# {'action': 'test'}
consumed == len(frame)
# True
Complete Example
from repod.protocol import encode, decode
# Encode a message
data = {"action": "chat", "message": "Hello!"}
encoded = encode(data)
# Decode the payload (skipping the 4-byte header)
decoded = decode(encoded[4:])
print(decoded)
# {'action': 'chat', 'message': 'Hello!'}