paulpas WebSocket Performance Optimization
Optimizes the data path of WebSocket connections for maximum throughput and minimum latency. Covers binary serialization (MessagePack, Protocol Buffers), per-message deflate compression negotiated during the handshake, message batching to reduce frame overhead, and payload size reduction techniques that directly shrink the bytes on the wire.
TL;DR Checklist
- Replace JSON serialization with MessagePack or Protobuf for structured messages — expect 3–5× smaller payloads
- Negotiate
permessage-deflateduring the WebSocket handshake with appropriate window size and compression level - Batch high-frequency small messages into single frames using time-based (≤10ms) or count-based (≤50 msg) aggregation
- Skip compression for already-compressed data (images, audio, encrypted payloads) — it adds CPU cost with zero benefit
- Profile serialization speed vs. payload size tradeoff: MessagePack is ~2× faster than JSON; Protobuf adds schema overhead but wins on repeated schemas
When to Use
- Building high-frequency trading tick data feeds where every microsecond and byte matters
- Real-time collaborative editing, gaming, or live dashboards sending hundreds of messages per second
- Mobile clients on expensive cellular networks where bandwidth conservation is critical
- Cross-region WebSocket links with measurable round-trip latency where batched reduces overhead
- Server-to-server real-time APIs where both endpoints control serialization format
When NOT to Use
- Simple chat or notification apps sending <50 messages per second — JSON adds negligible overhead at that rate
- Connections carrying binary blobs (images, audio, video frames) — these are already compressed; wrapping them in MessagePack or compressing with deflate wastes CPU
- Short-lived ephemeral connections (<1 minute) where handshake and compression setup overhead dominates
- When you need human-readable debug output on the wire — binary protocols sacrifice observability for efficiency
Core Workflow
Measure Current Baseline — Instrument your existing WebSocket path to measure payload size (compressed and uncompressed), serialization time, and messages per second. Checkpoint: Record
avg_payload_bytes,p99_latency_ms, andmsgs_per_secbefore making any changes.Choose Serialization Format — Evaluate MessagePack vs Protobuf based on your schema stability needs:
- Stable schemas with frequent updates → Protobuf for smallest payloads and fastest serialization
- Rapidly evolving schemas → MessagePack for runtime flexibility without code generation
- Maximum simplicity → JSON remains acceptable under ~50 msgs/sec with <1 KB messages
Negotiate Compression at Handshake — Enable
permessage-deflatein the WebSocket server and client configuration during the HTTP upgrade phase. Choose window size (4–15 KB for mobile, up to 65536 for server-to-server) and compression level (1–9). Checkpoint: VerifySec-WebSocket-Extensionsheader containspermessage-deflatein both the client request and server response.Implement Batching at Source — Group logically-related messages into batches before sending. Apply time-based batching (collect for ≤10 ms) or count-based batching (≤50 messages), whichever triggers first. Urgent messages bypass the batch queue. Checkpoint: Measure end-to-end latency impact — batching adds delay equal to the batch window; 10 ms batching should not violate your p99 latency target.
Benchmark and Iterate — Run A/B tests comparing JSON vs MessagePack vs Protobuf with and without compression. Measure throughput (msgs/sec), payload bytes, CPU utilization, and p99 latency. Checkpoint: Every optimization must improve at least one primary metric without degrading any other by more than 10%.
Implementation Patterns
Pattern 1: MessagePack Binary Protocol Over WebSockets
MessagePack serializes structured data into a compact binary format. Compared to JSON, it eliminates string keys from every message (keys are sent once in a schema), preserves numeric types (no "12345678901234567" string loss), and typically produces 3–5× smaller payloads for nested structures. Using Python's msgspec library for ~2× faster serialization than python-msgpack.
"""MessagePack binary serialization for WebSocket message exchange."""
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, asdict
from typing import Any
import msgspec
@dataclass
class TickData:
"""Market tick snapshot — a high-frequency trading data structure.
Attributes:
symbol: Trading pair identifier (e.g., "BTC/USD").
price: Last traded price in cents to avoid float precision issues.
size: Trade size in base currency units (integer micro-lots).
side: "buy" or "sell".
timestamp_ns: Unix epoch nanoseconds from exchange clock.
"""
symbol: str
price: int # cents, avoids float precision loss
size: int
side: str # "buy" | "sell"
timestamp_ns: int
@classmethod
def from_dict(cls, data: dict[str, Any]) -> TickData:
"""Construct a TickData from a JSON/dict payload.
Args:
data: Dict with keys matching TickData fields.
Returns:
Fully-constructed TickData instance.
"""
return cls(
symbol=data["symbol"],
price=int(data["price"]),
size=int(data["size"]),
side=data["side"],
timestamp_ns=int(data["timestamp_ns"]),
)
@dataclass
class OrderBookSnapshot:
"""Full order book snapshot for a trading pair.
Attributes:
symbol: Trading pair identifier.
bids: List of [price, size] pairs sorted descending by price.
asks: List of [price, size] pairs sorted ascending by price.
timestamp_ns: Server-side snapshot timestamp in nanoseconds.
"""
symbol: str
bids: list[list[int]]
asks: list[list[int]]
timestamp_ns: int
class MessagePackSerializer:
"""High-performance MessagePack serializer for WebSocket payloads.
Provides typed encode/decode methods that handle schema versioning,
message type dispatch, and benchmarking support for comparing against
JSON serialization.
Usage:
serializer = MessagePackSerializer()
tick = TickData(symbol="BTC/USD", price=4250000, size=100, side="buy", timestamp_ns=...)
raw = serializer.encode(tick) # → bytes
decoded = serializer.decode(raw, expected_type=TickData)
"""
def __init__(self, use_bin_type: bool = True) -> None:
"""Initialize the MessagePack serializer.
Args:
use_bin_type: If True (recommended), encode Python `bytes` as
MessagePack bin type instead of str. This preserves
binary data correctly when decoding on other languages.
"""
self._encoder = msgspec.msgpack.Encoder(use_bin_type=use_bin_type)
self._decoder = msgspec.msgpack.Decoder
def encode(self, obj: Any) -> bytes:
"""Serialize an object to MessagePack bytes.
Works with dataclasses, dicts, lists, and primitives.
For dataclasses, uses msgspec's native encoder which is ~3× faster
than json.dumps for structured data.
Args:
obj: The object to serialize. Must be a dataclass, dict, list,
or primitive type supported by msgspec.
Returns:
MessagePack-encoded bytes.
Raises:
msgspec.EncodeError: If the object contains unsupported types
(e.g., complex numbers, custom objects without
a registered encoder).
"""
return self._encoder.encode(obj)
def decode(self, data: bytes, expected_type: type | None = None) -> Any:
"""Deserialize MessagePack bytes into a Python object.
When expected_type is provided (a msgspec.Struct or dataclass),
the decoder performs strict type validation and field conversion
during deserialization — zero post-processing needed.
Args:
data: Raw MessagePack-encoded bytes.
expected_type: Optional type hint for strict decoding. If provided,
raises msgspec.DecodeError on mismatched fields.
Returns:
Decoded object. Type matches expected_type if specified,
otherwise returns dict/list/primitive as appropriate.
"""
decoder = self._decoder(expected_type) if expected_type else None
if decoder is not None:
return decoder.decode(data)
return msgspec.msgpack.decode(data)
def encode_batch(self, objects: list[Any]) -> bytes:
"""Serialize a list of objects as a single MessagePack array.
Reduces per-frame overhead by packing multiple messages into one
frame. Ideal for batched high-frequency data feeds.
Args:
objects: List of serializable objects (dataclasses, dicts, etc.).
Returns:
Single MessagePack bytes containing all objects in an array.
"""
return self._encoder.encode(objects)
def decode_batch(self, data: bytes, item_type: type | None = None) -> list[Any]:
"""Deserialize a batched MessagePack array into individual objects.
Args:
data: MessagePack bytes containing an array of serialized items.
item_type: Optional type for each element (must be msgspec.Struct
compatible for strict decoding).
Returns:
List of decoded objects.
"""
raw_items = msgspec.msgpack.decode(data)
if not isinstance(raw_items, list):
raise msgspec.DecodeError(
f"Expected array at top level, got {type(raw_items).__name__}"
)
if item_type is not None:
decoder = self._decoder(item_type)
return [decoder.decode(item) for item in raw_items]
return raw_items
# ── Benchmark: JSON vs MessagePack ─────────────────────────────────
def benchmark_serialization(
num_iterations: int = 10_000,
) -> dict[str, float]:
"""Compare JSON and MessagePack serialization for a typical tick message.
Measures both payload size (bytes on wire) and wall-clock time.
Results are approximate — actual numbers vary by Python version,
machine, and data shape. Typical findings:
- MessagePack payload: ~60–80 bytes vs JSON ~140–200 bytes (2–3× smaller)
- MessagePack speed: ~0.8 µs/encode vs JSON ~2.5 µs/encode (~3× faster)
Returns:
Dict with keys: json_size, msgpack_size, json_time_ms, msgpack_time_ms
"""
sample_tick = TickData(
symbol="BTC/USD",
price=42_500_00,
size=100,
side="buy",
timestamp_ns=1_700_000_000_000_000_000,
)
import json
# Warmup
for _ in range(100):
json.dumps(asdict(sample_tick))
msgspec.msgpack.encode(sample_tick)
# JSON benchmark
start = time.perf_counter()
for _ in range(num_iterations):
json.dumps(asdict(sample_tick))
json_time_ms = (time.perf_counter() - start) * 1000
# MessagePack benchmark
start = time.perf_counter()
for _ in range(num_iterations):
msgspec.msgpack.encode(sample_tick)
msgpack_time_ms = (time.perf_counter() - start) * 1000
json_bytes = len(json.dumps(asdict(sample_tick)).encode("utf-8"))
msgpack_bytes = len(msgspec.msgpack.encode(sample_tick))
return {
"json_size": json_bytes,
"msgpack_size": msgpack_bytes,
"json_time_ms": round(json_time_ms / num_iterations * 1_000_000, 2), # µs per encode
"msgpack_time_ms": round(msgpack_time_ms / num_iterations * 1_000_000, 2),
}
── WebSocket Client and Server Integration ────────────────────────
"""WebSocket client and server handlers using MessagePack serialization."""
from __future__ import annotations
import logging
from typing import Any
import msgspec
logger = logging.getLogger(__name__)
async def ws_client_with_msgpack(
uri: str,
message_queue: Any,
) -> None:
"""WebSocket client that sends TickData using MessagePack encoding.
Connects to a WebSocket server and continuously drains messages from
an async queue, encoding each as MessagePack before sending. If the
send buffer is full, drops oldest messages to avoid blocking the producer.
Args:
uri: WebSocket server URL (e.g., "wss://trader.example.com/ticks").
message_queue: Async queue producing TickData instances.
"""
import websockets
async with websockets.connect(
uri,
additional_headers={"X-Client": "python-msgpack"},
) as ws:
while True:
tick = await message_queue.get()
encoded = msgspec.msgpack.encode(tick)
try:
await ws.send(encoded)
except websockets.ConnectionClosed:
logger.error("Connection closed while sending tick for %s", tick.symbol)
break
async def ws_server_msgpack_handler(
websocket: Any,
) -> None:
"""WebSocket handler that decodes incoming MessagePack frames.
Demonstrates receiving typed MessagePack data from a client.
Uses msgspec's strict decoder to validate incoming fields —
rejects malformed messages before they reach application logic.
Args:
websocket: The websockets protocol object (async iterator over raw bytes).
"""
serializer = msgspec.msgpack
while True:
raw_frame = await websocket.recv()
try:
tick_dict = serializer.decode(raw_frame)
# Convert dict to typed TickData — type-safe, no KeyError risk
tick = TickData(
symbol=tick_dict["symbol"],
price=int(tick_dict["price"]),
size=int(tick_dict["size"]),
side=tick_dict["side"],
timestamp_ns=int(tick_dict["timestamp_ns"]),
)
process_tick(tick)
except (KeyError, msgspec.DecodeError) as exc:
await websocket.send(serializer.encode({
"error": "invalid_message",
"detail": str(exc),
}))
def process_tick(tick: TickData) -> None:
"""Process a validated tick — type-safe, no runtime key lookups.
Args:
tick: A fully-typed TickData instance from MessagePack decoding.
"""
# Direct attribute access — no dict['key'] indirection or KeyError risk
if tick.side == "buy" and tick.price > 43_000_00:
trigger_buy_signal(tick.symbol, tick.price)
def trigger_buy_signal(symbol: str, price: int) -> None:
"""Execute a buy signal for a given trading pair at the specified price.
In production this would integrate with the order management system,
placing a limit or market order based on strategy parameters. Here it
demonstrates the type-safe function call made after MessagePack decoding.
Args:
symbol: The trading pair that triggered the signal.
price: The triggering price in cents.
"""
# Log the signal with all available context for audit trail
logger.info("BUY SIGNAL: %s @ %d (%.2f)", symbol, price, price / 100)
# In production: order_manager.submit_order(symbol, side="buy", price=price)
Pattern 2: Protocol Buffers for Strongly-Typed Binary Serialization
Protocol Buffers provide schema-defined, language-independent binary serialization with automatic backward/forward compatibility. Unlike MessagePack's dynamic types, Protobuf encodes field numbers rather than names — the schema evolves without breaking existing clients that ignore unknown fields. This pattern shows .proto definition, Python code generation, and WebSocket integration.
"""Protocol Buffers schema definition and WebSocket serialization for real-time data."""
from __future__ import annotations
import asyncio
from dataclasses import dataclass
from typing import Any
# Generated by: protoc --python_out=. tick_data.proto
# The generated module provides TickMessage, OrderBookUpdate, and BatchMessage classes
# that map directly to .proto message definitions.
try:
# Import the generated protobuf module — created by running:
# pip install grpcio-tools
# protoc --python_out=. tick_data.proto
import tick_data_pb2 as proto_schema # type: ignore[import-not-found]
except ImportError:
# Stub module for development when .proto hasn't been compiled yet.
# In production, always compile with the latest .proto file.
class proto_schema: # type: ignore[no-redef]
class TickMessage:
symbol: str = ""
price: int = 0
size: int = 0
side: int = 0 # 0=buy, 1=sell
timestamp_ns: int = 0
@staticmethod
def FromString(data: bytes) -> "proto_schema.TickMessage":
raise NotImplementedError("Compile .proto with protoc first")
class OrderBookUpdate:
symbol: str = ""
side: int = 0
price: int = 0
size: int = 0
timestamp_ns: int = 0
@staticmethod
def FromString(data: bytes) -> "proto_schema.OrderBookUpdate":
raise NotImplementedError("Compile .proto with protoc first")
class BatchMessage:
messages: Any = []
@staticmethod
def FromString(data: bytes) -> "proto_schema.BatchMessage":
raise NotImplementedError("Compile .proto with protoc first")
# ── .proto Schema Definition ──────────────────────────────────────
#
# The .proto file that generates the above Python module:
#
# syntax = "proto3";
# package realtime;
# option py_generic_services = true;
#
# enum Side {
# SIDE_UNSPECIFIED = 0;
# SIDE_BUY = 1;
# SIDE_SELL = 2;
# }
#
# message TickMessage {
# string symbol = 1; // field number, NOT name, encoded in binary
# int64 price = 2; // stored as signed varint — no decimal point overhead
# uint64 size = 3;
# Side side = 4;
# fixed64 timestamp_ns = 5; // fixed-size encoding: always 8 bytes, fastest decoding
# }
#
# message OrderBookUpdate {
# string symbol = 1;
# Side side = 2;
# int64 price = 3;
# uint64 size = 4;
# fixed64 timestamp_ns = 5;
# }
#
# message BatchMessage {
# repeated TickMessage ticks = 1; // packed encoding: efficient for ordered arrays
# }
class ProtobufSerializer:
"""Serializes protobuf message objects to/from bytes over WebSockets.
Uses the compiled .proto-generated Python classes as both input and
output types, ensuring type safety throughout the serialization pipeline.
Schema Evolution Notes:
- Adding a new field with a higher field number is FORWARD COMPATIBLE:
old clients silently ignore the unknown field.
- Removing a field or changing its type breaks compatibility — only do
this on versioned schema updates (e.g., proto2 → proto3 migration).
- Never reuse field numbers within the same message type — reusing `1`
for a different purpose corrupts existing data when old clients parse.
Usage:
serializer = ProtobufSerializer()
tick_msg = proto_schema.TickMessage(
symbol="BTC/USD", price=4250000, size=100,
side=proto_schema.SIDE_BUY, timestamp_ns=time.time_ns(),
)
raw = serializer.encode(tick_msg)
decoded = serializer.decode(raw, proto_schema.TickMessage)
"""
def __init__(self) -> None:
"""Initialize the Protobuf serializer with no extra configuration.
Protobuf uses deterministic encoding by default (field ordering is
consistent), which is beneficial for caching and binary diffing.
All serialization state lives in the message objects themselves, so
the serializer is stateless and can be shared across tasks.
"""
def encode(self, message: Any) -> bytes:
"""Serialize a protobuf message to its compact binary wire format.
The output uses variable-length integer encoding (varint) for numeric
fields, making small numbers extremely compact: values <128 fit in 1
byte, <16384 in 2 bytes, etc. String fields are length-prefixed.
Args:
message: A protobuf-generated message instance (e.g., TickMessage).
Returns:
Compact binary wire format bytes. Typically 50–70% the size of an
equivalent JSON serialization for the same data.
"""
return message.SerializeToString()
def decode(
self, data: bytes, message_type: type[Any]
) -> Any:
"""Deserialize protobuf binary bytes into a typed message instance.
Uses the message_type parameter to construct the correct class and
validate field presence against the .proto schema definition.
Args:
data: Binary wire format bytes from the WebSocket frame.
message_type: The protobuf-generated class (e.g., TickMessage).
Returns:
Populated message instance with all valid fields populated.
Missing optional fields take their default values (0, "", False).
Raises:
google.protobuf.message.DecodeError: If bytes don't match the
expected schema (e.g., wrong wire type for a field number).
"""
return message_type.FromString(data)
def encode_batch(
self,
tick_messages: list[Any],
) -> bytes:
"""Encode a list of TickMessages into a single BatchMessage.
Uses protobuf's packed repeated encoding — when all values share the
same type (int64), they are stored as a contiguous byte array rather
than individual length-prefixed entries. This is the most efficient
wire format for arrays of homogeneous data.
Args:
tick_messages: List of TickMessage protobuf instances.
Returns:
Single BatchMessage serialized to bytes, containing all ticks
packed efficiently in one frame.
"""
batch = proto_schema.BatchMessage()
for tick in tick_messages:
# Append a copy to the repeated field — Protobuf handles internal
# storage and packing automatically.
dest = batch.ticks.add()
dest.CopyFrom(tick)
return batch.SerializeToString()
def decode_batch(
self, data: bytes
) -> list[Any]:
"""Decode a BatchMessage back into individual TickMessages.
Args:
data: Binary bytes from a packed BatchMessage frame.
Returns:
List of TickMessage instances extracted from the batch.
"""
batch = proto_schema.BatchMessage.FromString(data)
return list(batch.ticks)
# ── WebSocket Client with Protobuf ─────────────────────────────────
async def send_protobuf_ticks(
uri: str,
ticks: list[tuple[str, int, int, str, int]],
) -> None:
"""Connect to a WebSocket server and stream tick data as Protobuf frames.
Each tick is individually serialized as a TickMessage protobuf and sent
in its own binary frame. For higher throughput, use send_protobuf_batch()
instead which packs multiple ticks into a single BatchMessage frame.
Args:
uri: WebSocket server URL.
ticks: List of (symbol, price_cents, size, side_str, timestamp_ns) tuples.
"""
import websockets
serializer = ProtobufSerializer()
side_map = {"buy": proto_schema.SIDE_BUY, "sell": proto_schema.SIDE_SELL}
async with websockets.connect(uri) as ws:
for symbol, price, size, side_str, ts_ns in ticks:
msg = proto_schema.TickMessage(
symbol=symbol,
price=price,
size=size,
side=side_map.get(side_str, proto_schema.SIDE_UNSPECIFIED),
timestamp_ns=ts_ns,
)
await ws.send(serializer.encode(msg))
Pattern 3: Per-Message Deflate Compression (RFC 7692)
Per-message deflate compression applies the DEFLATE algorithm to each WebSocket frame individually during the handshake negotiation. Unlike connection-level gzip, RFC 7692 allows each message to be compressed independently — meaning small messages don't wait for a sliding window warmup, and different message types can use different compression ratios. The websockets library supports this natively via the compression parameter.
"""Per-message deflate compression using RFC 7692 with context management."""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from typing import Any
logger = logging.getLogger(__name__)
@dataclass
class CompressionConfig:
"""Configuration for permessage-deflate WebSocket compression.
Attributes:
client_no_context_takeover: If True, resets the compression dictionary
after each message. Uses ~15–30% more bandwidth
but eliminates memory growth from long-lived connections.
server_no_context_takeover: Same as above but for server-side compression.
compression_level: DEFLATE compression level 1–9 (1=fastest/lowest ratio,
9=slowest/highest ratio). Level 6 is a good default.
client_max_window_bits: Compression window size in bits. Lower values
use less memory but reduce compression ratio.
Range: 8 (256 bytes) to 15 (32 KB). Default: 15.
server_max_window_bits: Server-side window size, same range and semantics.
"""
client_no_context_takeover: bool = False
server_no_context_takeover: bool = False
compression_level: int = 6
client_max_window_bits: int = 15 # 32 KB window — good for text-heavy data
server_max_window_bits: int = 15
class PerMessageDeflateNegotiator:
"""Handles RFC 7692 permessage-deflate extension negotiation.
During the WebSocket handshake, both client and server exchange
Sec-WebSocket-Extensions headers declaring their compression capabilities.
This class builds those capability parameters and validates the server's
response against expected constraints.
Negotiation flow:
Client sends: Sec-WebSocket-Extensions: permessage-deflate;
client_no_context_takeover=0;
server_no_context_takeover=0;
client_max_window_bits=15;
server_max_window_bits=15
Server responds: permessage-deflate; server_no_context_takeover=0; ...
"""
def __init__(self, config: CompressionConfig) -> None:
"""Initialize negotiator with compression configuration.
Args:
config: The desired compression behavior and resource limits.
"""
self._config = config
def build_client_params(self) -> dict[str, str | bool | int]:
"""Build the capability parameters to send in the handshake request.
Returns:
Dict of parameter name → value pairs for the permessage-deflate
extension declaration on the client side.
"""
return {
"client_no_context_takeover": self._config.client_no_context_takeover,
"server_no_context_takeover": self._config.server_no_context_takeover,
"client_max_window_bits": self._config.client_max_window_bits,
"server_max_window_bits": self._config.server_max_window_bits,
}
def validate_server_response(
self,
server_params: dict[str, str | bool | int],
) -> tuple[bool, str]:
"""Validate the server's compression negotiation response.
Checks that the server accepted acceptable parameters and did not
request unreasonable resource usage (e.g., max_window_bits=15 on a
memory-constrained mobile device).
Args:
server_params: The parameters the server agreed to, parsed from
its Sec-WebSocket-Extensions header.
Returns:
Tuple of (is_acceptable, reason_if_not). If not acceptable,
raise ConnectionRefusedError with the reason.
"""
# Check that window sizes don't exceed our limits
server_window = int(server_params.get("client_max_window_bits", 15))
if server_window > self._config.client_max_window_bits:
return False, (
f"Server requested window_bits={server_window} but we limit "
f"to {self._config.client_max_window_bits}"
)
# Warn if server disabled our no_context_takeover preference
if not self._config.client_no_context_takeover and server_params.get(
"server_no_context_takeover"
):
logger.warning(
"Server enabled context takeover — memory grows with connection lifetime. "
"Use no-context-takeover for long-lived connections."
)
return True, ""
async def create_compressed_websocket_client(
uri: str,
config: CompressionConfig | None = None,
) -> Any:
"""Create a WebSocket client with permessage-deflate compression enabled.
Uses the `websockets` library's built-in compression support. When
`compress=0` is passed (default in websockets >= 10.0), the library
automatically negotiates permessage-deflate during handshake and applies
DEFLATE to every text and binary frame transparently.
Args:
uri: WebSocket server URL (wss:// or ws://).
config: Compression settings. If None, uses reasonable defaults.
Returns:
A configured websockets connection object ready for send/recv.
Raises:
websockets.InvalidHandshake: If the server does not support compression
and the client requires it.
"""
import websockets
cfg = config or CompressionConfig()
# Build the additional headers that declare our compression capabilities.
negotiator = PerMessageDeflateNegotiator(cfg)
params = negotiator.build_client_params()
async def compress_request(request: Any, path: str) -> Any:
"""Augment the handshake request with compression extension parameters."""
from websockets.http11 import Request
# The websockets library handles this automatically when compress is set,
# but this hook shows where the capability negotiation happens.
return request
async def compress_response(response: Any) -> Any:
"""Validate the server's compression response against our constraints."""
extensions_header = response.headers.get("Sec-WebSocket-Extensions", "")
if "permessage-deflate" in extensions_header.lower():
# Extract parameters from the header value for validation.
# In practice, websockets library handles this internally.
return response
logger.warning(
"Server did not negotiate permessage-deflate — all messages will be sent uncompressed"
)
return response
async with websockets.connect(
uri,
compress=cfg.compression_level, # Level 6 compression; pass 0 for auto-negotiate
additional_headers={
"Sec-WebSocket-Extensions": (
f'permessage-deflate; '
f'client_no_context_takeover={params["client_no_context_takeover"]}; '
f'server_no_context_takeover={params["server_no_context_takeover"]}; '
f'client_max_window_bits={params["client_max_window_bits"]}; '
f'server_max_window_bits={params["server_max_window_bits"]}'
)
},
) as ws:
return ws
# ── When NOT to Compress: BAD vs. GOOD ─────────────────────────────
async def bad_compress_everything(
websocket: Any,
) -> None:
"""❌ BAD — Compresses everything including already-compressed data.
Images, audio, encrypted payloads, and random binary blobs cannot be
compressed further (they're already at or near Shannon entropy limits).
Applying DEFLATE to these data types wastes CPU cycles and may even
increase the payload size by 2–5% due to deflate header overhead.
"""
import websockets
async with websockets.connect("wss://stream.example.com/binary", compress=9) as ws:
# Fetch a JPEG image from some endpoint, then send it over WebSocket
image_bytes = fetch_image_from_s3("photo.jpg") # Already compressed (JPEG)
await ws.send(image_bytes) # DEFLATE on JPEG = wasted CPU + slightly larger
audio_data = load_audio_file("voice.ogg") # Already compressed (Ogg Vorbis)
await ws.send(audio_data) # DEFLATE on OGG = wasted CPU + slightly larger
encrypted_payload = b"\x8f\x2a\xe1..." # Random bytes — cannot be compressed
await ws.send(encrypted_payload) # DEFLATE on random = ~3% bigger, no speedup
async def smart_compress(
websocket: Any,
data: bytes | str,
content_type: str,
) -> None:
"""✅ GOOD — Only compress text and JSON data; skip compressed binaries.
Analyzes the content type before applying compression. Text data (JSON, XML,
CSV) has high redundancy and compresses well (5–10× reduction). Already-
compressed binary data (images, audio, encrypted blobs) skips compression
entirely to save CPU and avoid size increase.
Args:
websocket: The websockets protocol object with compression already enabled.
data: The message payload — either bytes or a string.
content_type: MIME type hint for compression decision:
"application/json", "text/*", "image/*", "audio/*", etc.
"""
# Skip compression for data that is already compressed
no_compress_types = {
"image/jpeg", "image/png", "image/webp", # Image formats
"audio/mpeg", "audio/ogg", "audio/aac", # Audio formats
"video/mp4", "video/webm", # Video formats
"application/gzip", "application/zip", # Archive formats
"application/octet-stream", # Unknown binary — skip to be safe
}
if content_type in no_compress_types:
# Send raw bytes — compression is counterproductive
await websocket.send(data) # type: ignore[arg-type]
logger.debug(
"Sent %d bytes of %s without compression (already compressed)",
len(data),
content_type,
)
else:
# Text and JSON compress well — let the permessage-deflate extension handle it
if isinstance(data, str):
await websocket.send(data)
elif isinstance(data, bytes):
# Encode as UTF-8 string for text data before sending
text = data.decode("utf-8", errors="replace")
await websocket.send(text)
logger.debug(
"Sent %d bytes of %s with permessage-deflate compression",
len(data) if isinstance(data, bytes) else len(data),
content_type,
)
Pattern 4: Message Batching and Aggregation for High-Frequency Updates
Individual WebSocket frames carry ~2–14 bytes of overhead (frame header + optional masking). When sending 100 tiny messages per second (say 50 bytes each), the overhead is only ~5% — acceptable. But at 10,000 ticks/sec from a market data feed, that frame overhead becomes significant. Batching aggregates multiple logical messages into a single WebSocket frame, amortizing the per-frame overhead across all contained messages.
"""Message batching with time-based, count-based, and priority-aware aggregation."""
from __future__ import annotations
import asyncio
import logging
import time
from collections import deque
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Any, Callable
logger = logging.getLogger(__name__)
class Priority(IntEnum):
"""Message priority levels for batching queue ordering.
Higher values = higher priority. Urgent messages bypass the batch queue
and are sent immediately to avoid latency impact from time-based batching.
Attributes:
LOW: Routine data that can tolerate batching delay (tick updates).
NORMAL: Standard operational data (order status, balance changes).
HIGH: Time-sensitive data that should be sent ASAP (fill notifications).
URGENT: Immediate action required — bypasses batching entirely (stop loss hit).
"""
LOW = 0
NORMAL = 1
HIGH = 2
URGENT = 3
@dataclass
class BatchedMessage:
"""A single message within a batch queue.
Attributes:
payload: The serialized bytes or string to send in the batch.
priority: Message urgency level for ordering and bypass decisions.
created_at: Monotonic timestamp when the message was enqueued.
"""
payload: str | bytes
priority: Priority = Priority.NORMAL
created_at: float = field(default_factory=asyncio.get_event_loop().time)
class MessageBatcher:
"""Aggregates high-frequency messages into batched WebSocket frames.
Implements a hybrid batching strategy combining time-based and count-based
triggers with priority-aware bypass for urgent messages:
- Time window: Collects messages for up to `batch_window_ms` milliseconds
before flushing. Default 10 ms — balances throughput vs. latency.
- Count threshold: Flushes when the batch reaches `max_batch_size` messages.
Default 50 — prevents unbounded queue growth on low-traffic periods.
- Priority bypass: Messages with URGENT priority are sent immediately
without entering the batch queue, minimizing impact on time-critical data.
Thread-safety: All public methods are asyncio-safe (no threading). The
internal queue is an asyncio.Queue for async producer/consumer patterns.
Usage:
batcher = MessageBatcher(
batch_window_ms=10,
max_batch_size=50,
send_callback=websocket.send, # coroutine reference
)
await batcher.enqueue(tick_data) # Batches normally
await batcher.enqueue(urgent_msg, Priority.URGENT) # Sends immediately
await batcher.flush() # Forces immediate flush
"""
def __init__(
self,
batch_window_ms: int = 10,
max_batch_size: int = 50,
send_callback: Callable[[str | bytes], asyncio.Task[None]] | None = None,
flush_callback: Callable[[list[str | bytes]], asyncio.Task[None]] | None = None,
) -> None:
"""Initialize the message batcher.
Args:
batch_window_ms: Maximum milliseconds to hold messages in a batch
before flushing. Lower values reduce latency but
increase frame overhead. Typical range: 5–20 ms.
max_batch_size: Maximum number of messages per batch before flushing.
Prevents unbounded queue growth during low-traffic
periods where the time window is still active.
send_callback: Optional coroutine that sends a single message. If set,
urgent messages are sent via this callback directly.
flush_callback: Optional coroutine that sends a list of batched messages.
The batcher calls this to flush accumulated batches.
"""
self._window_ms = batch_window_ms
self._max_size = max_batch_size
self._send_callback = send_callback
self._flush_callback = flush_callback
# Internal queue of pending BatchedMessages
self._queue: deque[Ba
…(truncated)