Protocol Design & Serialization

# binary_protocol.py
import struct
import zlib
from dataclasses import dataclass
from enum import IntEnum

# Protocol constants
MAGIC = 0x4243      # 'BC' for ByteCode
VERSION = 1

class MessageType(IntEnum):
    PING = 1
    PONG = 2
    DATA = 3
    ERROR = 4

# Header format: big-endian, 12 bytes total
# ! = network byte order (big-endian)
# H = unsigned short (2 bytes) , magic
# B = unsigned char  (1 byte)  , version
# B = unsigned char  (1 byte)  , message type
# I = unsigned int   (4 bytes) , payload length
# I = unsigned int   (4 bytes) , checksum
HEADER_FORMAT = "!HBBII"
HEADER_SIZE = struct.calcsize(HEADER_FORMAT)   # = 12 bytes

@dataclass
class Message:
    msg_type: MessageType
    payload: bytes

def encode_message(msg: Message) -> bytes:
    """Encode a Message into wire-format bytes."""
    checksum = zlib.crc32(msg.payload) & 0xFFFFFFFF
    header = struct.pack(
        HEADER_FORMAT,
        MAGIC,
        VERSION,
        int(msg.msg_type),
        len(msg.payload),
        checksum,
    )
    return header + msg.payload

# Encode a DATA message
payload = b'{"sensor": "temp01", "value": 23.5, "unit": "C"}'
msg = Message(msg_type=MessageType.DATA, payload=payload)
wire_bytes = encode_message(msg)

print(f"Payload (JSON text): {len(payload)} bytes")
print(f"Wire frame (binary): {len(wire_bytes)} bytes (header={HEADER_SIZE})")
print(f"Header hex: {wire_bytes[:HEADER_SIZE].hex()}")
print(f"Full frame hex: {wire_bytes.hex()[:60]}...")

# binary_protocol_decode.py, continued
def decode_message(data: bytes) -> Message:
    """Decode wire-format bytes back into a Message. Raises on error."""
    if len(data) < HEADER_SIZE:
        raise ValueError(f"Too short: {len(data)} < {HEADER_SIZE}")

    magic, version, msg_type, payload_len, checksum = struct.unpack(
        HEADER_FORMAT, data[:HEADER_SIZE]
    )

    if magic != MAGIC:
        raise ValueError(f"Invalid magic: 0x{magic:04X} (expected 0x{MAGIC:04X})")
    if version != VERSION:
        raise ValueError(f"Unsupported version: {version}")

    payload = data[HEADER_SIZE:HEADER_SIZE + payload_len]
    actual_checksum = zlib.crc32(payload) & 0xFFFFFFFF

    if actual_checksum != checksum:
        raise ValueError(f"Checksum mismatch: {actual_checksum} != {checksum}")

    return Message(msg_type=MessageType(msg_type), payload=payload)

# Round-trip: encode then decode
decoded = decode_message(wire_bytes)
print(f"\nDecoded message type: {decoded.msg_type.name}")
print(f"Decoded payload: {decoded.payload.decode()}")

# Tampered frame, checksum catches it
tampered = bytearray(wire_bytes)
tampered[15] ^= 0xFF   # flip a bit in the payload
try:
    decode_message(bytes(tampered))
except ValueError as e:
    print(f"\nTampered frame rejected: {e}")

# msgpack_demo.py
import json
import msgpack
import time

data = {
    "user_id": 12345,
    "name": "Alice Johnson",
    "scores": [95, 87, 92, 78, 88],
    "metadata": {"role": "admin", "active": True, "level": 3},
}

# JSON serialization
json_bytes = json.dumps(data).encode("utf-8")

# MessagePack serialization
msgpack_bytes = msgpack.packb(data, use_bin_type=True)

print(f"JSON size:      {len(json_bytes)} bytes  → {json_bytes.decode()[:60]}...")
print(f"MessagePack:    {len(msgpack_bytes)} bytes  → {msgpack_bytes.hex()[:40]}...")
print(f"Size reduction: {(1 - len(msgpack_bytes)/len(json_bytes))*100:.1f}%")

# Deserialization
decoded = msgpack.unpackb(msgpack_bytes, raw=False)
print(f"\nDeserialized correctly: {decoded == data}")

# Speed comparison (1,000,000 iterations)
N = 1_000_000
start = time.perf_counter()
for _ in range(N):
    json.dumps(data)
json_time = time.perf_counter() - start

start = time.perf_counter()
for _ in range(N):
    msgpack.packb(data, use_bin_type=True)
msgpack_time = time.perf_counter() - start

print(f"\nJSON serialize 1M×:    {json_time:.3f}s")
print(f"MessagePack 1M×:       {msgpack_time:.3f}s")
print(f"MessagePack is {json_time/msgpack_time:.1f}× faster")