Topics

Topics implement a publish-subscribe communication pattern between nodes. A node emits a topic to publish messages, and other nodes consume that topic to receive them.

Use topics for continuous, unidirectional data streams such as sensor readings, camera frames, state updates, or any information that flows over time. Multiple consumers can listen to the same topic simultaneously.

Topics also underpin two higher-level constructs: contract implementation, where a topic’s contract lives in a standalone contract document that producers implement, and pairing, where two node instances exchange topics 1:1 in both directions over a two-role contract. For a side-by-side decision guide across every mechanism, see Choosing a communication pattern.

Emitting a topic

A node that publishes messages declares its topics under interfaces.topics.emits in its peppy.json5. Each topic defines a name, a qos_profile, and a message_format:

Python
Rust

{
  peppy_schema: "node/v1",
  manifest: {
    name: "uvc_camera",
    tag: "v1",
  },
  interfaces: {
    topics: {
      emits: [
        {
          name: "video_stream",
          qos_profile: "sensor_data",
          message_format: {
            header: {
              $type: "object",
              stamp: "time",
              frame_id: "u32"
            },
            encoding: "string",
            width: "u32",
            height: "u32",
            frame: {
              $type: "array",
              $items: "u8"
            }
          }
        }
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "uvc_camera"]
  },
}

{
  peppy_schema: "node/v1",
  manifest: {
    name: "uvc_camera",
    tag: "v1",
  },
  interfaces: {
    topics: {
      emits: [
        {
          name: "video_stream",
          qos_profile: "sensor_data",
          message_format: {
            header: {
              $type: "object",
              stamp: "time",
              frame_id: "u32"
            },
            encoding: "string",
            width: "u32",
            height: "u32",
            frame: {
              $type: "array",
              $items: "u8"
            }
          }
        }
      ],
    },
  },
  execution: {
    language: "rust",
    build_cmd: ["cargo", "build", "--release"],
    run_cmd: ["./target/release/uvc_camera"]
  },
}

The qos_profile controls delivery guarantees. Available profiles are:

sensor_data: optimized for high-frequency data where occasional drops are acceptable.
standard: balanced defaults suitable for most use cases. This is the default when qos_profile is omitted.
reliable: guarantees delivery at the cost of higher latency.
critical: strongest delivery guarantees for safety-critical data.

Publishing messages

After running peppy node sync, the code generator creates a module for each emitted topic under peppygen.emitted_topics (Python) / peppygen::emitted_topics (Rust). Each module exposes declare_publisher(node_runner), which returns a TopicPublisher, and build_message(...), which serializes the message fields into a payload. Declare the publisher once, then publish each message on it:

Python
Rust

import asyncio
import time

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.emitted_topics import video_stream


async def emit_frames(node_runner: NodeRunner):
    # Declare the publisher once; every publish below then reuses it.
    publisher = await video_stream.declare_publisher(node_runner)
    frame_id = 0
    while True:
        payload = video_stream.build_message(
            video_stream.MessageHeader(stamp=time.time(), frame_id=frame_id),
            "rgb8",
            640,
            480,
            bytes([1, 2, 3]),
        )
        await publisher.publish(payload)
        frame_id = (frame_id + 1) % (2**32)
        await asyncio.sleep(0.1)


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(emit_frames(node_runner))]


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use peppygen::emitted_topics::video_stream;
use peppygen::{NodeBuilder, Parameters, Result};
use std::time::Duration;

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        let node_runner_clone = node_runner.clone();
        tokio::spawn(async move {
            // Declare the publisher once; every publish below is then lock-free.
            let publisher = video_stream::declare_publisher(&node_runner_clone)
                .await
                .expect("failed to declare video_stream publisher");
            let mut frame_id = 0u32;
            loop {
                if let Ok(payload) = video_stream::build_message(
                    video_stream::MessageHeader {
                        stamp: std::time::SystemTime::now(),
                        frame_id,
                    },
                    "rgb8".to_owned(),
                    640,
                    480,
                    vec![1, 2, 3],
                ) {
                    let _ = publisher.publish(payload).await;
                }

                frame_id = frame_id.wrapping_add(1);
                tokio::time::sleep(Duration::from_secs_f64(0.1)).await;
            }
        });

        Ok(())
    })
}

build_message takes each field from the message_format in order (no node_runner argument) and returns the payload that publisher.publish sends; in Rust it returns a Result you unwrap, in Python it returns the payload directly and raises on invalid input. Nested objects become generated types whose fields match the object definition: a struct in Rust (video_stream::MessageHeader), a dataclass in Python (video_stream.MessageHeader).

A producer publishes a single stream regardless of how many consumers are listening, and it never knows about the bindings consumers use to address it. Starting the producer before any consumer (or after them) is equally valid; consumers attach when they show up.

For a topic with a simple message format:

{
  name: "message_stream",
  qos_profile: "sensor_data",
  message_format: {
    message: "string"
  }
}

build_message takes a single string argument:

Python
Rust

from peppygen.emitted_topics import message_stream

publisher = await message_stream.declare_publisher(node_runner)
payload = message_stream.build_message("hello world")
await publisher.publish(payload)

use peppygen::emitted_topics::message_stream;

let publisher = message_stream::declare_publisher(&node_runner).await?;
let payload = message_stream::build_message("hello world".to_owned())?;
publisher.publish(payload).await?;

Consuming a topic

A node that receives messages declares what it consumes under interfaces.topics.consumes. Dependencies are declared in manifest.depends_on and referenced by link_id in the interface:

Python
Rust

{
  peppy_schema: "node/v1",
  manifest: {
    name: "web_video_stream",
    tag: "v1",
    depends_on: {
      nodes: [
        { name: "uvc_camera", tag: "v1", link_id: "uvc_camera" },
      ]
    },
  },
  interfaces: {
    topics: {
      consumes: [
        {
          link_id: "uvc_camera",  // References depends_on.nodes[].link_id
          name: "video_stream",         // Topic name on that node
        },
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "web_video_stream"]
  },
}

{
  peppy_schema: "node/v1",
  manifest: {
    name: "web_video_stream",
    tag: "v1",
    depends_on: {
      nodes: [
        { name: "uvc_camera", tag: "v1", link_id: "uvc_camera" },
      ]
    },
  },
  interfaces: {
    topics: {
      consumes: [
        {
          link_id: "uvc_camera",  // References depends_on.nodes[].link_id
          name: "video_stream",         // Topic name on that node
        },
      ],
    },
  },
  execution: {
    language: "rust",
    build_cmd: ["cargo", "build", "--release"],
    run_cmd: ["./target/release/web_video_stream"]
  },
}

By default, the node add and node sync commands require the target node to already be in the node stack so the proper interfaces can be generated. If the target node is missing, you will see an error like:

Error: `web_video_stream:v1` depends on `uvc_camera:v1`, but it does not exist in the stack

Pass peppy node sync --include-repositories (-r) to let the daemon fall back to the repository cache for dependencies that aren’t in the stack.

Receiving messages

The code generator creates a module for each consumed topic under peppygen.consumed_topics (Python) / peppygen::consumed_topics (Rust). Every slot-backed module lives at <category>/<link_id>/<member>, keyed on the author’s link_id. The consumed module is therefore consumed_topics/<link_id>/<topic_name>, imported as from peppygen.consumed_topics.uvc_camera import video_stream (Python) / peppygen::consumed_topics::uvc_camera::video_stream (Rust); here the link_id is uvc_camera and the topic is video_stream. Call subscribe once to obtain a held Subscription, then await next for each message:

Python
Rust

import asyncio

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics.uvc_camera import video_stream


async def receive_frames(node_runner: NodeRunner):
    # Same subscribe() as every cardinality: it covers the slot's complete
    # bound set, which for `one` is exactly one producer. The yielded
    # ProducerRef is constant here, so the loop ignores it.
    subscription = await video_stream.subscribe(node_runner)

    async for _producer, frame in subscription:
        print(f"frame: {frame.width}x{frame.height}")


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(receive_frames(node_runner))]


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use peppygen::consumed_topics::uvc_camera::video_stream;
use peppygen::{NodeBuilder, Parameters, Result};

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        // Same subscribe() as every cardinality: it covers the slot's complete
        // bound set, which for `one` is exactly one producer. The yielded
        // ProducerRef is constant here, so the loop ignores it.
        let mut subscription = video_stream::subscribe(&node_runner).await?;

        while let Some((_, frame)) = subscription.next().await? {
            println!("frame: {}x{}", frame.width, frame.height);
        }

        Ok(())
    })
}

subscribe takes only node_runner (the node runner reference) and returns a Subscription covering the slot’s complete bound producer set: one producer-pinned wire subscription per bound producer, merged client-side behind the single subscribe() API. The shape is identical for every cardinality; only the size of the bound set changes. Awaiting subscription.next() yields:

a (producer, message) pair for each message (Ok(Some((producer, message))) in Rust), where producer is a ProducerRef carrying the publisher’s full (core_node, instance_id) wire identity, so the consumer can tell messages from different bound producers apart, and message is the deserialized message, with fields matching the topic’s message_format.
None (Ok(None) in Rust) once the node is shutting down and no queued message remains, or when every source has closed.
an error if a received payload fails to deserialize: Err(..) in Rust, a raised exception in Python. The error carries producer context and neither tears down the subscription nor shrinks the bound set.

Each underlying subscription holds an in-order buffer, so a message published between next calls is kept, not dropped; loop on next and nothing is lost in the gap between iterations. Message order is preserved independently per producer (no total ordering across producers is promised), and ready producers are merged fairly so one busy producer cannot indefinitely starve another. ProducerRef is Eq + Hash, so it keys per-producer state directly; to follow a single producer, filter on the yielded producer.

There are no call-site producer-targeting arguments at all. Which producers reach a slot is decided entirely by application bindings (see Bindings and routing below): a slot receives messages only from the producer instances bound to it, and each resolved binding carries the producer’s full (core_node, instance_id) wire address, so the slot can never match a producer that merely shares the instance_id on another core node. The producer segments of a keyexpr are never wildcarded: a federated router forwards traffic only for the explicitly bound producers, and every subscriber stays fully pinned (and auditable) in the zenoh admin space. That same full identity is what next returns with every message, so a consumer can pass it straight back into any producer-targeting call (a ProducerRef yielded by the slot’s own subscription is a member of the slot’s bound set by construction).

Every generated consumed topic, service, and action module also exposes a bound-producer accessor for its slot, returning the runtime-resolved, immutable producer binding in application declaration order. The accessor’s return type encodes exactly the guarantee launch validation established for the slot’s cardinality, instead of restating it in comments:

// Generated for every consumed module of a `cardinality: "one"` slot,
// replacing bound_producers() for that slot.
pub fn bound_producer(node_runner: &crate::NodeRunner) -> &ProducerRef;

// Generated for every consumed module of a `cardinality: "one_or_more"` slot:
// a never-empty ordered set whose first() is infallible.
pub fn bound_producers(node_runner: &crate::NodeRunner) -> NonEmptyProducers<'_>;

// Generated for every consumed module of a `cardinality: "zero_or_more"` slot:
// a plain, possibly empty slice, so the empty branch is forced by the type
// and is never dead code.
pub fn bound_producers(node_runner: &crate::NodeRunner) -> &[ProducerRef];

Python mirrors the split by name and annotation: a one slot generates bound_producer(node_runner) -> peppylib.ProducerRef, and the multi cardinalities generate bound_producers(node_runner) -> List[peppylib.ProducerRef] (documented never-empty for one_or_more).

Every module sharing the slot’s link_id returns the same set in the same order. The set is fixed when the node starts and is not live discovery: a producer disconnecting does not shrink it, trigger discovery, or cause automatic rebinding. A cardinality change is therefore visible at compile time in the consuming node: the accessor’s type changes and every call site that relied on the old guarantee must be revisited. This is intended; a cardinality flip is a semantic change, not a rebinding.

Receiving messages continuously

To receive messages continuously, subscribe once and loop on next. The held subscription keeps every message in arrival order, so the loop never misses a message published between iterations. Each message is processed in a separate task so that the receive loop is not blocked:

Python
Rust

import asyncio
import sys

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics.uvc_camera import video_stream


async def process_frame(producer, frame):
    print(
        f"got {frame.width}x{frame.height} frame "
        f"from {producer.core_node}/{producer.instance_id}"
    )


async def receive_frames(node_runner: NodeRunner):
    try:
        subscription = await video_stream.subscribe(node_runner)
    except Exception as e:
        print(f"Failed to subscribe: {e}", file=sys.stderr)
        return

    # The subscription is an async iterator; the loop ends when it closes.
    # Hold a reference to each task until it finishes, otherwise the event
    # loop may garbage-collect it mid-flight.
    tasks: set[asyncio.Task] = set()
    async for producer, frame in subscription:
        task = asyncio.create_task(process_frame(producer, frame))
        tasks.add(task)
        task.add_done_callback(tasks.discard)


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(receive_frames(node_runner))]


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use std::sync::Arc;
use peppygen::consumed_topics::uvc_camera::video_stream;
use peppygen::{NodeBuilder, NodeRunner, Parameters, Result};

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        tokio::spawn(receive_frames(node_runner));
        Ok(())
    })
}

async fn receive_frames(node_runner: Arc<NodeRunner>) {
    let mut subscription = match video_stream::subscribe(&node_runner).await {
        Ok(subscription) => subscription,
        Err(e) => {
            eprintln!("Failed to subscribe: {e}");
            return;
        }
    };

    loop {
        match subscription.next().await {
            Ok(Some((producer, frame))) => {
                tokio::spawn(async move {
                    println!(
                        "got {}x{} frame from {}/{}",
                        frame.width, frame.height, producer.core_node, producer.instance_id
                    );
                });
            }
            Ok(None) => break,
            Err(e) => {
                eprintln!("Error receiving frame: {e}");
                break;
            }
        }
    }
}

In Python the Subscription is also an async iterator: async for producer, message in subscription is equivalent to looping on next and stops cleanly once the subscription closes.

Consuming a multi-cardinality bound set

Nothing changes at the call site when the slot’s cardinality allows several producers: the same single subscribe() covers every bound producer, and the yielded ProducerRef tells the streams apart. ProducerRef is Eq + Hash, so it keys per-producer state directly:

Python
Rust

from peppygen.consumed_topics.camera import video_stream


async def count_frames(node_runner):
    # One subscription covering every producer bound to the `camera`
    # slot; the yielded producer tells per-camera streams apart.
    subscription = await video_stream.subscribe(node_runner)

    frames_per_camera: dict = {}
    async for producer, frame in subscription:
        frames_per_camera[producer] = frames_per_camera.get(producer, 0) + 1
        print(
            f"frame from {producer.instance_id}@{producer.core_node}: "
            f"{frame.width}x{frame.height} "
            f"(total from this camera: {frames_per_camera[producer]})"
        )

use peppygen::consumed_topics::camera::video_stream;
use peppygen::{NodeBuilder, Parameters, ProducerRef, Result};
use std::collections::HashMap;

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        // One subscription covering every producer bound to the `camera`
        // slot. The shape is identical for every cardinality; only the
        // size of the bound set changes.
        // `one`: the subscription covers exactly one producer.
        // `one_or_more`: the merged subscription covers at least one producer.
        // `zero_or_more`: the slot may be empty; the subscription then
        // yields nothing until the node stops.
        let mut subscription = video_stream::subscribe(&node_runner).await?;

        let mut frames_per_camera: HashMap<ProducerRef, u64> = HashMap::new();
        while let Some((producer, frame)) = subscription.next().await? {
            *frames_per_camera.entry(producer.clone()).or_insert(0) += 1;
            println!(
                "frame from {}@{}: {}x{} (total from this camera: {})",
                producer.instance_id,
                producer.core_node,
                frame.width,
                frame.height,
                frames_per_camera[&producer],
            );
        }
        Ok(())
    })
}

To follow a single camera instead, filter on the yielded producer (bound_producers() lists the members in binding order).

Bindings and routing

Routing is entirely consumer-side. Producers publish unconditionally and are binding-agnostic; only the consumer’s depends_on slots and the application bindings that fill them decide which producer instance reaches which slot. (Pairing topics are the one exception: a pairing slot is routed by the pair itself, established with --link at instance start, not by a binding.)

A binding takes the form KEY: VALUE, where KEY is a link_id declared in the consumer’s depends_on and VALUE selects the slot’s target producer instance(s). Each bound target creates a private one-way channel from that producer instance to the consumer’s slot: a multi-cardinality slot expands into N independent application-selected bound edges sharing the same link_id. How many targets a slot takes is the slot’s declared cardinality; there is no wildcard fallback and no discovery, only explicitly bound producers.

In a launcher / stack config:

{
  source: { local: "./consumer" },
  instances: [{
    instance_id: "my_consumer",
    links: { uvc_camera: "my-camera-instance" },
  }],
}

or, when launching a single node during development:

peppy node run --link uvc_camera@my-camera-instance .

Dependency cardinality

Every depends_on.contracts and depends_on.nodes entry may declare a cardinality, constraining how many producers the application may bind to the slot:

depends_on: {
  contracts: [
    { name: "uvc_camera", tag: "v1", link_id: "camera", cardinality: "one_or_more" },
  ],
}

Cardinality	Valid unique targets	Omitted binding
`one` / omitted	exactly 1	error
`one_or_more`	1 or more	error
`zero_or_more`	0 or more	empty set

The binding value’s shape mirrors the slot’s cardinality. A one slot takes a scalar string only, exactly as above. A one_or_more or zero_or_more slot takes an array of instance ids; for zero_or_more the empty array [] is a valid definition, equivalent to omitting the binding entirely:

links: {
  camera: ["front_camera", "rear_camera"],
}

Any other combination is an error: an array on a one slot (single-element arrays included), a scalar on a multi slot, and an empty array on a one_or_more slot (cardinality unmet). The CLI mirrors the array form by repetition, which accumulates on a multi slot and stays a hard error on a one slot; the empty set has no flag spelling, omission is it:

peppy node run --link camera@front_camera --link camera@rear_camera .

Cardinality applies to a consumer’s dependency slot; it does not permit undeclared or unbound launched node instances, and it constrains the application-configured bound set at startup, not runtime availability: a producer disconnecting does not shrink the set, trigger discovery, or cause automatic rebinding. depends_on.pairings entries have no cardinality (a pairing is strictly 1:1 and expresses absence with its optional flag; a cardinality key on a pairing entry is a manifest error).

The consuming API is uniform across cardinalities in everything except the bound-producer accessor: topics subscribe to the slot’s complete bound set and yield the producing ProducerRef with every message, and services and actions require the caller to select one explicit, membership-checked member of that set, whatever its size. The accessor itself is cardinality-typed: one generates the singular, infallible bound_producer(), one_or_more generates bound_producers() returning a never-empty set whose first() needs no unwrap, and zero_or_more generates bound_producers() returning a possibly empty slice, so the empty branch is forced by the type. For a zero_or_more slot bound to nothing, the subscription simply yields nothing until the node stops.

Per-message routing

For every message a producer publishes, each consumer instance applies one rule on its own slots: the message is delivered to every slot whose bound set names the producer, and to no other slot. A producer named by no binding is ignored by the consumer entirely; there is no wildcard fallback. Wildcard-subscribe-and-filter-in-process is deliberately not offered: a wildcard expresses interest in every same-namespace producer of the contract, so a federated router would forward traffic from unbound producers across the mesh, and the bound set would stop being auditable from the zenoh admin space.

Worked example: `openarm01_backbone`

A consumer that wires two specific depth cameras to dedicated wrist slots:

{
  manifest: {
    name: "openarm01_backbone",
    tag: "v1",
    depends_on: {
      contracts: [
        { name: "depth_camera", tag: "v1", link_id: "wrist_left_camera" },
        { name: "depth_camera", tag: "v1", link_id: "wrist_right_camera" },
      ],
    },
  },
  // ...
}

And the launcher that binds it:

{
  deployments: [
    { source: { name: "depth_camera:v1" }, instances: [
      { instance_id: "left_cam" },
      { instance_id: "right_cam" },
    ]},
    { source: { name: "openarm01_backbone:v1" }, instances: [
      { instance_id: "backbone_inst_1", links: {
        wrist_left_camera: "left_cam",
        wrist_right_camera: "right_cam",
      }},
    ]},
  ],
}

Three contract statements follow from this manifest:

A frame from left_cam arrives only on the wrist_left_camera slot.
A frame from right_cam arrives only on the wrist_right_camera slot.
If the wrist_right_camera binding line were removed, the launch would be rejected: every declared slot except zero_or_more must be bound. A producer named by no binding is simply ignored, but a declared one or one_or_more slot with no binding is an error, not a silent gap.

Dedicated slots (one link_id per camera, as here) and one multi-cardinality slot (cameras: ["left_cam", "right_cam"]) are both valid designs. Dedicated slots give each producer a distinct role in code; a multi slot treats the producers as an N-of-a-kind set behind one API.

Validator rules

The launcher validator runs these checks before the stack starts:

Every KEY must name a declared slot, and every declared slot must resolve. A --link KEY@VALUE (or launcher links: entry) whose KEY matches no depends_on.{nodes,contracts} link_id is rejected; there are no free-form keys. In the other direction, a declared one or one_or_more slot the bindings leave out fails the launch with one error per unfulfilled slot; a zero_or_more slot with no binding resolves to the empty set.
The value’s shape must match the slot’s cardinality. A one slot takes a scalar only (an array is rejected, single-element and empty included), a multi slot takes an array only (a scalar is rejected), and an empty array meets only zero_or_more (on one_or_more it is a cardinality-unmet error). Repeated --link KEY@… flags carry no shape and are checked by count: more than one occurrence on a one slot is rejected. Duplicate targets within one slot are rejected rather than deduplicated.
Every target must satisfy the slot, checked per bound instance. For node slots, each target producer’s (node_name, node_tag) must equal the slot’s declared pair (BindingTargetMismatch otherwise). For contract slots, each target’s manifest.implements must include the slot’s (name, tag) (BindingContractNotImplemented otherwise); see Contract implementation for the implementation rules.
Stack-wide instance_id uniqueness. Every instance_id must be unique across the entire stack, not just within a (node_name, node_tag) group. The --link syntax names producers by instance_id, so reusing one across two different node kinds would make the binding ambiguous.
Bindings are stamped with the daemon’s core_node. The --link KEY@instance_id syntax names producers by instance_id alone, but the wire addresses producers by the full (core_node, instance_id) pair, since instance_id is only unique within one stack. The validator stamps the launching daemon’s core_node into every resolved binding, so the runtime never matches on instance_id alone: a slot subscribes with both wire fields set. A producer on another core node that happens to share an instance_id can never feed a bound slot. Bound-set member order is the application declaration order (launcher array order / CLI flag occurrence order), preserved verbatim into the runtime configuration.

Message format types

The message_format supports the following field types (see the message format reference for the full type system, including aliases and nesting rules):

Type	Rust type	Python type
`"bool"`	`bool`	`bool`
`"u8"`	`u8`	`int`
`"u16"`	`u16`	`int`
`"u32"`	`u32`	`int`
`"u64"`	`u64`	`int`
`"i8"`	`i8`	`int`
`"i16"`	`i16`	`int`
`"i32"`	`i32`	`int`
`"i64"`	`i64`	`int`
`"f32"`	`f32`	`float`
`"f64"`	`f64`	`float`
`"string"`	`String`	`str`
`"bytes"`	`Vec<u8>`	`bytes`
`"time"`	`std::time::SystemTime`	`float` (Unix epoch seconds)

Fields can also use complex types:

Object: a nested struct in Rust, a generated dataclass in Python:

header: {
  $type: "object",
  stamp: "time",
  frame_id: "u32"
}

Array: a variable-length list (Vec<T> in Rust, list[T] in Python; an array of u8 maps to Vec<u8> / bytes):
```
frame: {
  $type: "array",
  $items: "u8"
}
```
Fixed-length array: an array with a known size ([T; N] in Rust, a list[T] that must hold exactly N items in Python):
```
position: {
  $type: "array",
  $items: "f32",
  $length: 3
}
```
Optional: a field that may be absent (Option<T> in Rust, T | None in Python):
```
error_msg: {
  $type: "string",
  $optional: true
}
```