Skip to content
Peppy

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

Section titled “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:

{
  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"]
  },
}

The qos_profile controls delivery guarantees. Available profiles are:

  • sensor_data: optimised 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

Section titled “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:

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()

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:

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)

Consuming a topic

Section titled “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:

{
  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"]
  },
}

Receiving messages

Section titled “Receiving messages”

The code generator creates a module for each consumed topic under peppygen.consumed_topics (Python) / peppygen::consumed_topics (Rust). The module name is <link_id>_<topic_name> based on the link_id field; in this case uvc_camera_video_stream. Call subscribe once to obtain a held Subscription, then await next for each message:

import asyncio


from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics import uvc_camera_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 uvc_camera_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()

subscribe takes:

  • 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

Section titled “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:

import asyncio
import sys


from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics import uvc_camera_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 uvc_camera_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()

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

Section titled “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:

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 = camera_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

Section titled “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 --pair 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",
    bindings: { uvc_camera: "my-camera-instance" },
  }],
}

or, when launching a single node during development:

Terminal window
peppy node run --bind uvc_camera@my-camera-instance .

Dependency cardinality

Section titled “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" },
  ],
}
CardinalityValid unique targetsOmitted binding
one / omittedexactly 1error
one_or_more1 or moreerror
zero_or_more0 or moreempty 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:

bindings: {
  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:

Terminal window
peppy node run --bind camera@front_camera --bind 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

Section titled “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

Section titled “Worked example: openarm01_backbone”

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

openarm01_backbone/peppy.json5
{
  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:

peppy_launcher.json5
{
  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", bindings: {
        wrist_left_camera: "left_cam",
        wrist_right_camera: "right_cam",
      }},
    ]},
  ],
}

Three contract statements follow from this manifest:

  1. A frame from left_cam arrives only on the wrist_left_camera_video_stream slot.
  2. A frame from right_cam arrives only on the wrist_right_camera_video_stream slot.
  3. 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

Section titled “Validator rules”

The launcher validator runs these checks before the stack starts:

  1. Every KEY must name a declared slot, and every declared slot must resolve. A --bind KEY@VALUE (or launcher bindings: 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.
  2. 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 --bind 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.
  3. 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.
  4. 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 --bind syntax names producers by instance_id, so reusing one across two different node kinds would make the binding ambiguous.
  5. Bindings are stamped with the daemon’s core_node. The --bind 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

Section titled “Message format types”

The message_format supports the following field types:

TypeRust typePython type
"bool"boolbool
"u8"u8int
"u16"u16int
"u32"u32int
"u64"u64int
"i8"i8int
"i16"i16int
"i32"i32int
"i64"i64int
"f32"f32float
"f64"f64float
"string"Stringstr
"bytes"Vec<u8>bytes
"time"std::time::SystemTimefloat (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
    }