Messages¶

Source: cortex.messages.base, cortex.messages.standard

A message is a @dataclass that inherits from Message. Registration, fingerprinting, and serialization are automatic.

Anatomy of a message¶

classDiagram
    class Message {
        +fingerprint() int
        +to_bytes() bytes
        +to_frames() list
        +from_bytes(data) tuple
        +from_frames(frames) tuple
        +decode(bytes) tuple [static]
        -_build_header()
        -_field_names() tuple
        -_field_values() list
        -_next_sequence() int
    }
    class MessageHeader {
        +fingerprint: int
        +timestamp_ns: int
        +sequence: int
        +to_bytes() bytes
        +from_bytes(data) MessageHeader
        +size() int
    }
    class MessageType {
        +register(cls)
        +get(fingerprint) type
        +get_all() dict
    }
    Message ..> MessageHeader : emits
    Message ..> MessageType : auto-registers on subclass

Defining a custom message¶

from dataclasses import dataclass
import numpy as np
from cortex.messages.base import Message

@dataclass
class JointTrajectory(Message):
    timestamp: float
    positions: np.ndarray   # shape (N,)
    velocities: np.ndarray  # shape (N,)
    frame_id: str = ""

The class registers into MessageType._registry by fingerprint at import time and gains:

• JointTrajectory.fingerprint() — 64-bit ID.
• msg.to_frames() / JointTrajectory.from_frames(frames) — transport path.
• msg.to_bytes() / JointTrajectory.from_bytes(data) — legacy single-blob path.
• Message.decode(blob) — class dispatch via fingerprint registry.

Sequence numbering¶

Per-publisher monotonic counter, attached to each message's header. A class-level fallback counter on Message._sequence_counter covers direct to_bytes/to_frames use outside a Publisher (tests, ad-hoc serialization).

Built-in messages¶

Encode / decode lifecycle¶

flowchart LR
    A[User builds dataclass] --> B[Publisher.publish]
    B --> C[message.to_frames]
    C --> D[[ZMQ multipart send]]
    D --> E[[ZMQ multipart recv]]
    E --> F[Message.from_frames]
    F --> G[user callback msg, header]

See also¶

• Concept: message wire format
• Concept: fingerprinting
• Tutorial: custom messages