Zenoh Programming in Rust

This book is a practical guide to building distributed systems with Eclipse Zenoh and Rust.

It targets Zenoh 1.4.0 and the Rust 2024 edition. Each chapter introduces a concept, explains its design rationale, and demonstrates it with working code.

Scope

Prerequisites

• Rust toolchain installed (rustup)
• Familiarity with Rust ownership and async/await
• Basic understanding of pub/sub messaging

Source Code

All examples in this book are available at https://github.com/kydos/zenoh-book.

Introduction to Zenoh

This chapter defines what Zenoh is, where it comes from, how it is structured internally, and when it is the right tool for the job.

What Zenoh Is

Zenoh is a unified protocol for publish/subscribe, querying, and distributed storage. All three operations share a single hierarchical keyspace, so a publisher, a subscriber, and a queryable storage node can all operate on the same named resource without any translation layer between them.

The three fundamental operations are:

• Put — write a value to a key. Any subscriber matching that key receives the sample. Any storage node covering that key persists it.
• Subscribe — receive a stream of samples as they are published to keys matching a key expression.
• Get — issue a query against a key expression and receive replies from queryables and storage nodes that hold matching data.

The three corresponding abstractions are:

• Publisher — declares intent to write to a key expression and sends samples efficiently.
• Subscriber — declares interest in a key expression and receives samples from publishers.
• Queryable — registers a handler for queries arriving on a key expression and sends replies back to the requester.

Zenoh is not a broker protocol. There is no central message broker that all traffic passes through. Sessions form direct peer-to-peer connections when possible, and route through infrastructure nodes (routers) only when topology requires it. This design lets Zenoh run on microcontrollers with tens of kilobytes of RAM, on embedded Linux gateways, and on datacenter servers, all using the same wire protocol and the same API surface.

Zenoh Genesis

Around 2009–2010, while working on extremely large-scale systems spanning military, aerospace, and smart city deployments, it became clear that no existing protocol could cover the full spectrum from microcontroller to datacenter. Each protocol solved part of the problem and failed somewhere else on the stack.

DDS provided location transparency for data in motion and a data-centric model, but could not scale down to constrained hardware and lost its location-transparency guarantees the moment data was stored — at which point applications were forced back to centralized cloud storage.

CoAP was effective for integrating small devices with web applications but remained inherently client/server and cloud-centric.

MQTT introduced a lightweight broker-based pub/sub model suited to cloud-dependent applications. Its broker topology, however, created what could be called the MQTT paradox: two devices on the same local network still route every message through a broker running on a cloud server thousands of kilometres away, adding latency and creating a single point of failure.

The result of combining these protocols to cover a full system was what could be called the Digital Frankenstein era: large-scale cloud-to-microcontroller systems duct-taped together from a series of protocol stacks, each covering one network segment, with no unified semantics across the whole.

Zenoh was designed to close that gap. The goal was a single protocol that works efficiently from microcontroller to datacenter, with no topological constraints, and that provides unified abstractions for data in motion (pub/sub), data at rest (distributed queries), and location-transparent computation.

The name itself encodes the design intent. It references Zeno of Elea — the pre-Socratic philosopher known for his paradoxes of infinity — and Zenon of Citium, the founder of Stoicism. It also stands for ZEro Network OverHead, reflecting the protocol’s efficiency-first design philosophy.

Core Architecture

Zenoh defines three operational modes for a session:

• Peer — the default mode. A peer participates in multicast scouting to discover other peers on the same network segment, then establishes direct connections. No infrastructure node is required. Suitable for local networks and development environments.
• Client — a thin mode for constrained nodes. A client connects to one or more routers and delegates all routing to them. Clients do not route for other nodes.
• Router — a full infrastructure node. Routers connect to other routers and serve clients. They forward messages between network segments and can host storage nodes.

Transport flexibility is a first-class design goal. The same Zenoh session can communicate over:

• TCP (reliable, ordered, connection-oriented)
• UDP (unreliable, used for multicast scouting and datagram transports)
• QUIC (reliable, multiplexed, with built-in TLS)
• TLS (TCP with certificate-based authentication)
• Shared memory (SHM, zero-copy on the same host)
• Serial (for embedded links)

The transport layer is selected by the endpoint URI scheme (tcp/, quic/, tls/, shm/). Multiple transports can be active simultaneously on a single session.

Wire overhead is minimal by design: a Zenoh data message requires as few as 4 bytes of framing overhead. The session is the entry point to everything. A session is opened with a configuration, and all publishers, subscribers, and queryables are declared on that session. All transport, routing, and resource management flow through it.

When to Use Zenoh

Use Zenoh when:

• You need publish/subscribe, queryable storage, and request/reply in a single protocol. Zenoh covers all three without requiring separate middleware stacks or translation bridges between them.
• You need to span microcontrollers, embedded gateways, and cloud servers with one protocol and one API. Zenoh-Pico runs on bare-metal targets with tens of kilobytes of RAM; the same wire protocol connects to a datacenter router.
• You need zero-copy shared-memory transport for high-throughput, low-latency communication between processes on the same host — camera pipelines, LiDAR streams, ML inference data.
• You need to integrate other protocols. Zenoh is designed to act as a protocol backbone: connectors exist for DDS, MQTT, ROS 2, and REST, allowing existing systems to participate in a Zenoh keyspace without being rewritten.
• You need to span edge, fog, and cloud across heterogeneous network segments — WiFi, cellular, Ethernet, serial — with automatic topology-aware routing and no broker single point of failure.
• You are building a system where data must be stored, queried, and streamed with consistent semantics. Zenoh’s queryable storage model avoids the split between a messaging layer and a separate database layer.
• You need efficient operation under intermittent connectivity. Zenoh routers buffer and forward when endpoints reconnect; storages answer queries from local replicas when upstream is unavailable.

Key Terminology

Getting Started

This chapter walks through installing the required tools, setting up a Rust project, and writing the first working Zenoh programs.

Prerequisites

Zenoh 1.4.0 requires Rust stable 1.75 or later and the 2024 edition. Install Rust via rustup:

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

After installation, verify the toolchain:

rustc --version
cargo --version

Zenoh’s async API is built on Tokio. The tokio crate is a direct dependency of your application, not just of Zenoh internally — you need it in your Cargo.toml to use #[tokio::main] and to drive the async runtime.

Installing zenohd

zenohd is the Zenoh router daemon. In peer mode, two processes on the same local network can discover and communicate directly without it. Once you need to cross network segments, connect machines on different subnets, or host storage nodes, you run zenohd as the infrastructure node.

macOS

brew install zenoh

Ubuntu and Debian

Add the Eclipse Zenoh apt repository, then install:

echo "deb [trusted=yes] https://download.eclipse.org/zenoh/debian-repo/ /" \
  | sudo tee /etc/apt/sources.list.d/zenoh.list
sudo apt update
sudo apt install zenoh

Windows

Download the zip archive from the Zenoh releases page, extract it, and add the directory to your PATH.

Verify the installation

zenohd --version

The output shows the version string, e.g., zenohd v1.4.0.

Project Setup

Create a new Cargo project:

cargo new zenoh_hello
cd zenoh_hello

Edit Cargo.toml to specify the 2024 edition and add the required dependencies:

[package]
name = "zenoh_hello"
version = "0.1.0"
edition = "2024"

[dependencies]
zenoh = "1.4.0"
tokio = { version = "1", features = ["full"] }

The "full" feature set for Tokio enables all runtime components. For production use you can trim this to the specific features you need (rt-multi-thread, macros, etc.), but "full" is the correct starting point.

Hello Zenoh

The minimal Zenoh program opens a session in default peer mode and puts a single value:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");

    let session = zenoh::open(Config::default()).await.unwrap();
    session.put("demo/hello", "Hello, Zenoh").await.unwrap();
    println!("Published.");
}

zenoh::init_log_from_env_or("error") initialises the Zenoh logger. The argument is the fallback log level when the ZENOH_LOG environment variable is not set. Passing "error" keeps output quiet during normal operation.

Config::default() produces a peer-mode configuration with multicast scouting enabled. No router endpoint needs to be configured.

session.put(key, value) is a convenience method. The key is any string that forms a valid Zenoh key (no wildcards). The value is anything that implements Into<ZBytes> — string literals qualify automatically.

Run the program:

cargo run

The output is:

Published.

The put succeeds even if no subscriber is running. Zenoh delivers to whichever subscribers exist at the time of publication; there is no error if the set is empty.

First Pub/Sub Pair

To see data actually received, run a subscriber in one terminal before starting the publisher. Create a second binary or use [[bin]] sections in Cargo.toml. The subscriber example:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");

    let session = zenoh::open(Config::default()).await.unwrap();
    let subscriber = session.declare_subscriber("demo/**").await.unwrap();

    while let Ok(sample) = subscriber.recv_async().await {
        let payload = sample
            .payload()
            .try_to_string()
            .unwrap_or_else(|e| e.to_string().into());
        println!("{}: {}", sample.key_expr(), payload);
    }
}

declare_subscriber registers interest in all keys under demo/. The ** wildcard matches any number of path segments (see Chapter 3 for the full wildcard syntax).

recv_async().await yields the next sample from the internal channel. The loop runs until the channel is closed, which happens when the subscriber is dropped or the session closes.

try_to_string() attempts to interpret the payload bytes as UTF-8. It returns a Result; the unwrap_or_else branch converts decode errors to their string representation so the loop always prints something readable.

Running the pair

Terminal 1 — start the subscriber first so it is registered before the publisher sends:

cargo run --bin subscriber

Terminal 2 — run the publisher:

cargo run --bin publisher

Terminal 1 prints:

demo/hello: Hello, Zenoh

Running with zenohd

In default peer mode, zenohd is optional. Peers on the same multicast-reachable network segment discover each other automatically via UDP multicast on 224.0.0.224:7446 and establish direct TCP connections.

To use client mode — for example, to connect two machines that cannot reach each other by multicast — start zenohd on one machine:

zenohd

Then configure each client session to connect to the router’s TCP endpoint. Full configuration is covered in Chapter 4. The short form using programmatic configuration:

#![allow(unused)]
fn main() {
let mut config = Config::default();
config.insert_json5("mode", r#""client""#).unwrap();
config.insert_json5("connect/endpoints", r#"["tcp/192.168.1.10:7447"]"#).unwrap();
let session = zenoh::open(config).await.unwrap();
}

Replace 192.168.1.10 with the address of the machine running zenohd.

Troubleshooting

Publisher and subscriber do not communicate in peer mode

Multicast scouting requires that UDP multicast traffic is allowed between the two processes. Check:

• The firewall on each machine permits UDP on 224.0.0.224:7446.
• Both processes are on the same network segment (multicast does not cross routers by default).
• No VPN or network namespace is isolating the processes.

Set ZENOH_LOG=debug to see scouting messages:

ZENOH_LOG=debug cargo run --bin subscriber

Look for lines containing scouting or scout to confirm discovery is proceeding.

Build fails with edition errors

Zenoh 1.4.0 requires edition = "2024" in Cargo.toml. If you see errors about unstable features or edition-gated syntax, verify that the [package] section specifies edition = "2024" and that your Rust toolchain is 1.75 or later.

Session opens but put/subscribe calls fail immediately

Unwrapping on zenoh::open and on put will panic on error. During development this is acceptable; the panic message includes the underlying error. If you see a panic at session open, the most common cause is a misconfigured endpoint string — verify that the address format matches the scheme (e.g., tcp/host:port, not tcp://host:port).

The Zenoh Data Model

This chapter defines the data structures and naming conventions that underlie every Zenoh operation: keys, key expressions, selectors, payloads, encodings, timestamps, and samples.

Keys

A key is a /-separated sequence of UTF-8 path segments that uniquely identifies a resource in the Zenoh keyspace. Examples:

robot1/sensors/lidar
plant/zone3/temperature
vehicle/fleet/truck42/gps/position

Formation rules:

• No leading slash: sensors/lidar, not /sensors/lidar.
• No trailing slash: sensors/lidar, not sensors/lidar/.
• No empty segments: sensors//lidar is invalid.
• Segments cannot contain *, $, ?, or #. Those characters are reserved for key expressions and selectors.

A key is used in write operations (put, delete) and as the identity of a stored resource. It represents a single, concrete resource — not a pattern.

Key Expressions

A key expression (KE) is a string that matches one or more keys. Key expressions are used when declaring subscribers, queryables, and publishers, and when issuing get queries. A plain key with no wildcards is also a valid key expression that matches exactly itself.

Wildcard Forms

* — single-segment wildcard

Matches exactly one non-empty path segment. The segment may contain any character except /.

robot1/sensors/*

Matches: robot1/sensors/lidar, robot1/sensors/camera, robot1/sensors/imu

Does not match: robot1/sensors/lidar/scan (two segments where one is expected)

$* — substring wildcard

Matches any substring within a single segment. It can appear at the start, end, or middle of a segment alongside literal characters.

robot$*/sensors

Matches: robot1/sensors, robot42/sensors, robot_arm/sensors

Does not match: sensors (no first segment), robot1/motor/sensors (extra segment)

A standalone $* in a segment position is equivalent to *.

** — multi-segment wildcard

Matches zero or more consecutive path segments, including the empty case (matching the position between two slashes, or the start/end of the path).

robot1/**/temperature

Matches: robot1/temperature, robot1/sensors/temperature, robot1/sensors/internal/temperature

** by itself matches any key.

Canon Form

Zenoh normalises key expressions to a canonical form automatically. You do not need to apply these rules manually, but understanding them helps avoid writing redundant patterns:

• **/** reduces to ** (two consecutive multi-segment wildcards are one)
• **/* becomes */** (the single-segment wildcard floats to the left of the multi-segment)
• $*$* reduces to $* (two adjacent substring wildcards merge)
• A standalone $* in a full-segment position is equivalent to *

Attempting to use a malformed key expression — for example, one with adjacent ** separated by a slash — will produce an error at declaration time, before any data is exchanged.

Intersection and Inclusion

Two key expressions intersect if there exists at least one key that matches both. A key expression A includes key expression B if every key matching B also matches A. These relations are used internally by Zenoh’s routing layer to determine which subscribers receive which publications, and which queryables receive which queries.

Selectors

A selector extends a key expression with an optional query parameter string:

key/expression?param1=val1;param2=val2

The query string uses URL-query-string syntax with ; as the separator between parameters. Selectors appear in session.get() calls:

#![allow(unused)]
fn main() {
let replies = session.get("robot1/sensors/**?encoding=json").await.unwrap();
}

The key expression portion is used for routing — it determines which queryables receive the query. The query string is forwarded to the queryable as-is. The queryable interprets the parameters according to its own logic; Zenoh does not enforce any meaning on the parameter names or values.

When no query string is needed, pass the key expression string directly to session.get(). The ? and everything after it are optional.

ZBytes

ZBytes is the universal payload container in Zenoh. Every sample carries its data as a ZBytes value. The type holds an arbitrary sequence of bytes and provides no built-in schema — the structure of the bytes is defined by the application and communicated via the Encoding field.

Creating ZBytes

Standard Rust types convert to ZBytes via From/Into:

#![allow(unused)]
fn main() {
// From a string slice
let bytes: ZBytes = "Hello, Zenoh".into();

// From a String
let s = String::from("sensor reading");
let bytes: ZBytes = s.into();

// From a byte slice
let raw: &[u8] = &[0x01, 0x02, 0x03];
let bytes: ZBytes = raw.into();
}

Reading ZBytes

For UTF-8 text payloads, use try_to_string():

#![allow(unused)]
fn main() {
let text = sample.payload().try_to_string()
    .unwrap_or_else(|e| e.to_string().into());
println!("{text}");
}

try_to_string() returns a Result<Cow<str>, _>. On success it gives a borrowed string if the bytes are already UTF-8, or an error if they are not.

For structured data — serialisation and deserialisation with z_serialize and z_deserialize — see Chapter 9.

What to watch out for

ZBytes is not self-describing. A receiver cannot determine the structure of the payload from the bytes alone. Always pair ZBytes with an accurate Encoding value so receivers can make an informed decision about how to decode it.

Encoding

Encoding is advisory metadata attached to every sample. It tells the receiver what format the payload bytes are in, using a MIME-type-like identifier. The receiver is free to ignore it, but well-behaved applications check the encoding before attempting to decode.

Standard encoding constants

Custom encodings

#![allow(unused)]
fn main() {
use zenoh::bytes::Encoding;

let enc = Encoding::new("application/x-my-format");
}

The string is arbitrary. By convention, use type/subtype format following MIME conventions. Custom encodings with the same string on both sides of a session are sufficient for application-level agreement.

Setting encoding on a put

#![allow(unused)]
fn main() {
use zenoh::bytes::Encoding;

session
    .put("sensors/temperature", payload)
    .encoding(Encoding::APPLICATION_JSON)
    .await
    .unwrap();
}

Reading encoding from a sample

#![allow(unused)]
fn main() {
let enc = sample.encoding();
println!("Encoding: {enc}");
}

Timestamps

Every Zenoh sample carries a timestamp generated by the originating session. The timestamp is a 64-bit Hybrid Logical Clock (HLC) value with approximately 3.5 nanosecond resolution, combined with the UUID of the session that created it.

The HLC advances monotonically even across clock adjustments on the originating machine, and the combination of the HLC value plus the session UUID guarantees a unique total order across the distributed system. No two samples from the same session share the same timestamp; no two samples from different sessions share the same (timestamp, uuid) pair.

Timestamps are used by storage plugins to perform conflict resolution: when two writes arrive at a storage node for the same key, the one with the later HLC timestamp wins.

Accessing the timestamp

#![allow(unused)]
fn main() {
if let Some(ts) = sample.timestamp() {
    println!("Timestamp: {ts}");
}
}

The timestamp is Option<Timestamp> because samples created directly by application code without going through the full Zenoh stack may lack one. In practice, samples received from a live network always carry a timestamp.

Sample

A Sample is the unit of data delivery in Zenoh. It is what subscribers receive and what queryables return as replies. A sample carries:

The kind field distinguishes a data write (Put) from a deletion (Delete). Delete samples carry a key expression and a timestamp but no meaningful payload. Subscribers and storage nodes use the kind to determine whether to store or remove the resource.

A typical subscriber handler pattern:

#![allow(unused)]
fn main() {
while let Ok(sample) = subscriber.recv_async().await {
    match sample.kind() {
        SampleKind::Put => {
            let payload = sample.payload().try_to_string()
                .unwrap_or_else(|e| e.to_string().into());
            println!("PUT  {}: {payload}", sample.key_expr());
        }
        SampleKind::Delete => {
            println!("DEL  {}", sample.key_expr());
        }
    }
}
}

The Admin Space

Zenoh exposes internal runtime state under the reserved @/ key prefix. The admin space is a live, queryable view into the Zenoh infrastructure.

What is available

• @/router/<router-id>/info — identity and version of the router.
• @/router/<router-id>/links — active transport links and their statistics.
• @/router/<router-id>/subscribers — declared subscribers visible to this router.
• @/router/<router-id>/queryables — declared queryables visible to this router.
• @/router/<router-id>/config/... — runtime configuration (if write access is enabled).

Reading admin data

#![allow(unused)]
fn main() {
let replies = session.get("@/router/**").await.unwrap();
while let Ok(reply) = replies.recv_async().await {
    if let Ok(sample) = reply.result() {
        let payload = sample.payload().try_to_string()
            .unwrap_or_else(|e| e.to_string().into());
        println!("{}: {payload}", sample.key_expr());
    }
}
}

Runtime configuration changes

When the admin space is enabled with write permissions (see Chapter 4), you can PUT to @/router/<id>/config/... to modify router settings at runtime without restarting the daemon. Use this capability carefully in production: an incorrect config write can disrupt routing for all connected sessions.

Sessions and Configuration

This chapter covers how to open and close a Zenoh session, how to supply configuration from a file or programmatically, and what the three operational modes mean in practice.

Opening a Session

A Zenoh session is the top-level object through which all operations flow. Open one with zenoh::open:

#![allow(unused)]
fn main() {
use zenoh::Config;

let session = zenoh::open(Config::default()).await.unwrap();
}

zenoh::open takes a Config value and returns a Result<Session, ZError>. The await drives the async handshake: scouting, transport binding, and any router connections configured in the config complete before the future resolves.

The returned Session is an Arc-backed handle. Cloning it is cheap and gives another handle to the same underlying session. All clones share the same transport state, resource declarations, and lifecycle.

Drop the session to close it, or call session.close().await for explicit control (see the Graceful Shutdown section below).

Default Configuration

Config::default() produces a configuration that:

• Sets the mode to peer.
• Enables multicast scouting on udp/224.0.0.224:7446.
• Listens for incoming connections on any available TCP port.
• Disables the admin space.

This configuration requires no setup and works immediately on a single machine or a local network where multicast is not blocked. It is the correct starting point for development and single-machine testing.

When the default configuration is not appropriate — for example, when multicast is unavailable, when you need to connect across subnets, or when you need TLS — you supply an explicit configuration.

Configuration from File

Zenoh uses JSON5 as its configuration file format. JSON5 is a superset of JSON that allows single-line comments (//), trailing commas, and unquoted keys. Load a config file with:

#![allow(unused)]
fn main() {
let config = Config::from_file("path/to/config.json5").unwrap();
let session = zenoh::open(config).await.unwrap();
}

A representative config file covering the most common fields:

{
  // Operational mode: "peer", "client", or "router"
  mode: "client",

  connect: {
    // Endpoints this session connects to (client or peer mode)
    endpoints: ["tcp/192.168.1.10:7447"]
  },

  listen: {
    // Endpoints this session listens on (router or peer mode)
    endpoints: ["tcp/0.0.0.0:7447"]
  },

  adminspace: {
    enabled: true,
    permissions: {
      read: true,
      write: false   // set true to allow runtime config changes
    }
  }
}

Key fields

• mode — controls routing behaviour. Covered in detail in the Connection Modes section.
• connect.endpoints — the list of remote endpoints to dial on session open. Each endpoint is a URI string with the scheme determining the transport: tcp/, udp/, quic/, tls/.
• listen.endpoints — the list of local addresses to bind and accept connections on. In peer mode this is optional; the runtime picks an ephemeral port when omitted.
• adminspace.enabled — exposes internal state and configuration under the @/ keyspace (see Chapter 3 for the admin space query pattern).
• adminspace.permissions.write — when true, allows PUT operations to @/router/<id>/config/... to change settings at runtime.

What to watch out for

Config file paths are resolved relative to the current working directory of the process, not relative to the config file itself. Use absolute paths in production deployments to avoid surprises when the working directory changes.

Programmatic Configuration

When you need to construct configuration at runtime — reading from environment variables, command-line arguments, or a service registry — use Config::default() as a base and mutate it with insert_json5:

#![allow(unused)]
fn main() {
let mut config = Config::default();
config.insert_json5("mode", r#""client""#).unwrap();
config.insert_json5("connect/endpoints", r#"["tcp/127.0.0.1:7447"]"#).unwrap();
let session = zenoh::open(config).await.unwrap();
}

insert_json5 takes a path (using / as a separator to navigate nested keys) and a JSON5-encoded value as a string. The value string must be valid JSON5 for the target type:

• A string value: r#""client""# (note the inner quotes)
• An array: r#"["tcp/127.0.0.1:7447"]"#
• A boolean: "true" or "false"
• A number: "7447"

insert_json5 returns a Result. Unwrapping is fine during development; in production, propagate the error.

Combining file and programmatic config

You can load a base configuration from a file and then override specific fields programmatically:

#![allow(unused)]
fn main() {
let mut config = Config::from_file("base.json5").unwrap();
// Override the connect endpoint from an environment variable
let endpoint = std::env::var("ZENOH_ENDPOINT").unwrap_or_else(|_| "tcp/127.0.0.1:7447".into());
let json_endpoint = format!(r#"["{endpoint}"]"#);
config.insert_json5("connect/endpoints", &json_endpoint).unwrap();
let session = zenoh::open(config).await.unwrap();
}

Connection Modes

Peer

The default mode. A peer:

• Binds a local listen endpoint (TCP port, chosen automatically if not specified).
• Sends and receives multicast UDP scouting messages to discover other peers on the same network segment.
• Establishes direct unicast connections to discovered peers.
• Performs local routing: if peer A and peer B are both connected to peer C, peer C forwards messages between A and B.

Peer mode requires no pre-existing infrastructure. Two processes on the same machine in peer mode discover each other within milliseconds of starting. On a LAN, peers across machines discover each other through multicast scouting.

Peer mode is appropriate for: development, local robotics deployments, and any scenario where all participants are on the same multicast-reachable network segment.

Client

A client:

• Does not bind any listen endpoint.
• Does not perform multicast scouting (by default).
• Connects to one or more routers specified in connect.endpoints.
• Delegates all routing to the connected routers.

Client mode is appropriate for: constrained devices that should not act as routing hops, processes behind NAT that cannot accept incoming connections, and any node that should be a pure consumer or producer without infrastructure responsibility.

A client session cannot communicate with another client session directly. Both must be connected to a common router.

Router

A router:

• Binds listen endpoints and accepts connections from clients and other routers.
• Participates in router-to-router routing to forward messages across network segments.
• Can host storage plugins and admin space access.

You typically run zenohd rather than embedding a router in application code, but the router mode is available programmatically when needed (for example, in integration tests or embedded gateway applications).

Session Info

After opening a session, session.info() returns a handle to runtime identity information:

#![allow(unused)]
fn main() {
use zenoh::session::ZenohId;

let info = session.info();
let zid: ZenohId = info.zid().await;
let routers: Vec<ZenohId> = info.routers_zid().await.collect();
let peers: Vec<ZenohId> = info.peers_zid().await.collect();

println!("My ZID: {zid}");
println!("Connected routers: {routers:?}");
println!("Connected peers: {peers:?}");
}

• zid() returns the UUID of this session. The ZID is stable for the lifetime of the session and is used as the session identifier in timestamps and admin space paths.
• routers_zid() returns an async stream of the ZIDs of all routers this session is currently connected to. The .collect() call drains the stream into a Vec.
• peers_zid() returns an async stream of peer ZIDs (in peer mode).

session.info() is useful for diagnostics, for logging the session identity on startup, and for constructing admin space query paths that reference a specific router.

Runtime Configuration via Admin Space

When the admin space is open with write permissions, you can change router configuration at runtime by putting JSON5-encoded values to @/router/<router-id>/config/<path>:

#![allow(unused)]
fn main() {
// Enable a new listen endpoint at runtime (example only — not all fields support hot reload)
let router_zid = session.info().routers_zid().await.next().await.unwrap();
let config_key = format!("@/router/{router_zid}/config/listen/endpoints");
session
    .put(&config_key, r#"["tcp/0.0.0.0:7448"]"#)
    .await
    .unwrap();
}

Not every configuration field supports hot reload. Fields that affect transport binding or routing topology typically require a restart. Consult the Zenoh router documentation for the current list of runtime-configurable fields.

Use this capability with caution in production. An incorrect write can disrupt routing for all sessions connected to the router. Restrict write access to the admin space unless runtime reconfiguration is an explicit operational requirement.

Graceful Shutdown

A Zenoh session closes when it is dropped. Rust’s ownership model ensures the session is always closed, but “always” here means “at the point the owning variable goes out of scope.” In a long-running async application, that point can be far from where you logically want the session to stop.

For explicit control, call session.close():

#![allow(unused)]
fn main() {
session.close().await.unwrap();
}

close() consumes the session handle, performs an orderly teardown — undeclaring resources, draining send queues, and closing transport links — and returns a Result. Awaiting it ensures the teardown completes before your program proceeds.

Publishers and subscribers before close

Publishers and subscribers are cleaned up automatically when the session closes, whether by drop or by explicit close(). You do not need to call undeclare() on each entity before closing the session; the session close handles it.

If you need to undeclare a specific entity earlier in the session’s lifetime — for example, to stop receiving on a subscriber without closing the whole session — call subscriber.undeclare().await.unwrap(). This is covered in more detail in the chapters on publishers and subscribers.

What to watch out for

If you hold a Session clone and the original is dropped, the session stays open because the Arc reference count is still positive. The session closes only when all clones are dropped (or when close() is called on any clone, which closes the underlying session for all handles).

Publishers and Put

This chapter covers publishing data: one-shot put, the delete operation, and declared publishers for repeated writes.

One-Shot Put

session.put(key_expr, payload) publishes a single value to the network and returns. No resource is pre-allocated before the call. This is appropriate for infrequent writes where per-call setup overhead is acceptable.

What it is: A single publish operation that sends one sample to all matching subscribers and storages.

When to use it: Use put when writes are infrequent — configuration updates, one-time events, or scripts that publish once and exit.

How to use it:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    session.put("demo/example/zenoh-rs-put", "Put from Rust").await.unwrap();
}

You can attach encoding metadata to the payload with .encoding(). The encoding is informational — zenoh does not transcode data. Receivers can use it to select a deserializer.

use zenoh::{bytes::Encoding, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    session
        .put("demo/temperature", "23.5")
        .encoding(Encoding::TEXT_PLAIN)
        .await
        .unwrap();
}

What to watch out for: Each call to session.put() performs routing resolution on every invocation. If your publish rate exceeds a few Hz, declare a publisher instead to avoid repeated overhead.

Delete

session.delete(key_expr) publishes a Sample with kind = SampleKind::Delete. Subscribers receive the sample and can act on it. Storages remove the entry for that key. The key is not required to have a prior Put — a delete on an unknown key is well-defined and simply results in a no-op at storages that do not hold it.

What it is: A publish operation that signals removal of a key rather than the presence of a value.

When to use it: Use delete when you want to signal that a resource no longer exists — for example, when a device goes offline or a configuration entry is removed.

How to use it:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    session.delete("demo/example/zenoh-rs-put").await.unwrap();
    session.close().await.unwrap();
}

What to watch out for: Subscribers that do not inspect sample.kind() will receive delete samples as regular deliveries. If your subscriber only handles put data, check the kind before accessing the payload, since a delete sample carries no meaningful payload.

Declared Publishers

session.declare_publisher(key_expr) pre-declares the publisher with the routing fabric. The routing path is negotiated once at declaration time. Subsequent writes use the pre-established path, eliminating per-write routing resolution. Use declared publishers for high-frequency writes.

What it is: A publisher handle bound to a fixed key expression, with routing pre-negotiated at declaration time.

When to use it: Use a declared publisher when writing at more than a few Hz, when you want publisher-side QoS options, or when you want the routing fabric to pre-establish paths before the first write.

How to use it:

use std::time::Duration;
use zenoh::{bytes::Encoding, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let publisher = session
        .declare_publisher("demo/example/zenoh-rs-pub")
        .await
        .unwrap();

    for idx in 0..u32::MAX {
        tokio::time::sleep(Duration::from_secs(1)).await;
        let buf = format!("[{idx:4}] Pub from Rust");
        publisher
            .put(buf)
            .encoding(Encoding::TEXT_PLAIN)
            .await
            .unwrap();
    }
}

What to watch out for: The declared publisher is bound to a single key expression. It does not support wildcard key expressions in the key it publishes to — each put call on the publisher sends to the exact key declared. The publisher handle must remain in scope for the duration of use; dropping it undeclares the publisher and releases routing resources.

Publisher Options

QoS options are set at declaration time via builder methods. They apply to every put call made through that publisher.

What it is: Per-publisher quality-of-service settings that control delivery semantics.

When to use it: Set these when the default best-effort, drop-on-congestion behavior is not acceptable for your use case.

How to use it:

• reliability(Reliability::Reliable) — acknowledged delivery; the default is BestEffort
• priority(Priority::RealTime) — one of seven priority levels from RealTime down to Background
• congestion_control(CongestionControl::Block) — block the publisher when the network is congested; the default is Drop
• express(true) — bypass batching to minimize latency at the cost of throughput

use zenoh::{qos::{CongestionControl, Priority, Reliability}, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let publisher = session
        .declare_publisher("control/cmd")
        .reliability(Reliability::Reliable)
        .priority(Priority::RealTime)
        .congestion_control(CongestionControl::Block)
        .await
        .unwrap();
    publisher.put("stop").await.unwrap();
}

What to watch out for: CongestionControl::Block causes the put call to block the current async task when the network is saturated. In an async context this can delay other tasks sharing the same executor thread. Full QoS details are in chapter 10.

Matching Listener

A declared publisher can observe whether any matching subscribers currently exist. This allows a publisher to avoid expensive serialization work when no one is listening.

What it is: A callback that fires whenever the set of matching subscribers changes — transitioning from zero to non-zero, or non-zero to zero.

When to use it: Use this to gate expensive payload construction, or to log connectivity state in long-running services.

How to use it:

#![allow(unused)]
fn main() {
publisher
    .matching_listener()
    .callback(|status| {
        if status.matching() {
            println!("Publisher has matching subscribers.");
        } else {
            println!("Publisher has no matching subscribers.");
        }
    })
    .background()
    .await
    .unwrap();
}

What to watch out for: The matching status reflects the routing fabric’s view of subscriber interest. A subscriber on a remote node may take a short time to propagate through the routing layer. Do not treat the matching status as an instantaneous guarantee.

put() vs declare_publisher()

Use session.put() for occasional writes where per-call setup overhead is negligible — scripts, command-line tools, or infrequent state updates. Use session.declare_publisher() when writing at more than a few Hz, when you want publisher-side QoS pre-configured at declaration time, or when you want the routing fabric to pre-establish paths before the first write arrives.

The declared publisher also enables the matching listener feature. One-shot put has no equivalent mechanism for observing subscriber presence.

Subscribers

This chapter covers receiving data: declaring subscribers, processing samples, using wildcard key expressions, and the pull model with ring channels.

Declaring a Subscriber

session.declare_subscriber(key_expr) registers interest in all puts and deletes matching the key expression. It returns a Subscriber handle backed by a FIFO channel by default. Incoming samples are queued and consumed by calling recv_async() on the handle.

What it is: A subscription handle that delivers all samples published to matching key expressions.

When to use it: Use a subscriber whenever your application needs to react to data published by other nodes — sensor readings, telemetry, commands, or any event-driven data flow.

How to use it:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let subscriber = session
        .declare_subscriber("demo/example/**")
        .await
        .unwrap();

    while let Ok(sample) = subscriber.recv_async().await {
        let payload = sample
            .payload()
            .try_to_string()
            .unwrap_or_else(|e| e.to_string().into());
        println!(
            "{} {} '{}'",
            sample.kind(),
            sample.key_expr().as_str(),
            payload
        );
    }
}

What to watch out for: The loop above exits when recv_async() returns Err, which happens when the subscriber handle is dropped or the session is closed. If you hold the session and subscriber in the same scope, both must remain alive for the loop to continue running.

Sample Fields

Each received Sample carries the following fields.

What it is: The complete unit of data delivery in zenoh — a key, a value, and metadata.

When to use it: Inspect sample fields whenever you need more than the payload: filtering by key, checking timestamps for ordering, or branching on put vs delete.

How to use it:

• key_expr() — the concrete key of this sample; no wildcards appear here even if the subscription used wildcards
• payload() — the ZBytes value carrying the raw bytes
• timestamp() — optional HLC timestamp set by the origin; None if the publisher did not attach one
• encoding() — the declared encoding; use this to select a deserializer
• kind() — SampleKind::Put or SampleKind::Delete
• attachment() — optional auxiliary bytes attached by the publisher

#![allow(unused)]
fn main() {
while let Ok(sample) = subscriber.recv_async().await {
    if let Some(ts) = sample.timestamp() {
        println!("ts={ts}");
    }
    println!("kind={} key={}", sample.kind(), sample.key_expr().as_str());
}
}

What to watch out for: timestamp() returns None unless the publisher explicitly set one or unless a zenoh router with timestamping enabled inserted it. Do not assume timestamps are always present.

Wildcard Subscriptions

Key expressions in subscriptions match all publications whose key intersects the pattern. Wildcards follow the zenoh key expression syntax: * matches a single chunk, ** matches any number of chunks.

What it is: A subscription whose key expression contains wildcards, causing it to receive samples from multiple publishers that each use concrete keys.

When to use it: Use wildcards when you want to receive from a class of publishers without naming each one explicitly — for example, all sensors under a prefix, or all topics at a given depth.

How to use it:

• demo/example/** receives everything under that prefix
• */sensors/* receives any two-chunk key with sensors as the middle segment
• demo/+/temperature uses + as a single-segment wildcard (synonym for * in many contexts)

Wildcards in the subscription do not appear in the received sample.key_expr(). That field always contains the concrete key the publisher used.

What to watch out for: Overly broad wildcards such as ** at the root match every key expression in the system. This is rarely what you want and can generate unexpected traffic. Use the narrowest pattern that covers your use case.

Callback-Based Subscriber

Pass a closure with .callback() for fire-and-forget processing. The callback runs in the zenoh receiver thread and should complete quickly. Use .background() to keep the subscriber alive without holding an explicit handle.

What it is: A subscriber variant where incoming samples are dispatched to a user-supplied closure instead of a channel.

When to use it: Use callbacks when you do not need backpressure control, when the processing is lightweight, and when you want to avoid writing a receive loop.

How to use it:

#![allow(unused)]
fn main() {
let _subscriber = session
    .declare_subscriber("demo/example/**")
    .callback(|sample| {
        println!("received: {}", sample.key_expr().as_str());
    })
    .background()
    .await
    .unwrap();

// Keep the program alive
std::thread::park();
}

.background() keeps the subscriber alive as long as the session is alive, without holding an explicit handle.

What to watch out for: The callback runs synchronously on the receiver thread. Blocking inside the callback — for example with a mutex, a slow I/O call, or an async runtime — stalls the delivery of subsequent samples. If processing is non-trivial, hand off work to a separate task inside the callback.

Pull Model: Ring Channel

By default the subscriber uses a FIFO. If the consumer is slower than the producer, the FIFO grows unbounded. Use RingChannel to cap memory and automatically drop the oldest samples when the channel is full.

What it is: An alternative channel handler that bounds the subscriber’s internal queue and discards old entries when new ones arrive and the ring is full.

When to use it: Use RingChannel for sampling use cases where only the latest value matters — telemetry dashboards, health monitors, and other consumers that can tolerate dropping stale data.

How to use it:

use std::time::Duration;
use zenoh::{handlers::RingChannel, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let subscriber = session
        .declare_subscriber("demo/example/**")
        .with(RingChannel::new(3))  // keep at most 3 samples
        .await
        .unwrap();

    loop {
        match subscriber.recv_async().await {
            Ok(sample) => {
                let payload = sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into());
                println!("Pulled: {}", payload);
                tokio::time::sleep(Duration::from_secs(5)).await;
            }
            Err(e) => {
                println!("Error: {e}");
                return;
            }
        }
    }
}

What to watch out for: Samples dropped by the ring channel are silently discarded — there is no notification that a drop occurred. If your application must process every sample without loss, use the default FIFO channel and ensure the consumer keeps up with the producer.

Subscriber Lifecycle

The subscriber is active as long as its handle is held. Dropping the handle undeclares the subscriber. Resources on remote routers are released and the routing fabric stops forwarding matching publications to this node.

What it is: The relationship between the Rust handle and the active subscription registered with the routing fabric.

When to use it: Understand the lifecycle to avoid two common mistakes: dropping the handle too early (subscription disappears) or holding it too long (resources are not released after the subscription is no longer needed).

How to use it:

Always keep the subscriber handle in scope for the duration you need it:

#![allow(unused)]
fn main() {
let subscriber = session.declare_subscriber("demo/data/**").await.unwrap();
// subscriber is active here
process_data(&subscriber).await;
// subscriber is dropped here; resources are released
}

Use .background() to tie the subscriber lifetime to the session rather than to a local variable:

#![allow(unused)]
fn main() {
session
    .declare_subscriber("demo/data/**")
    .callback(|s| println!("{}", s.key_expr().as_str()))
    .background()
    .await
    .unwrap();
// subscriber remains active until the session is closed
}

What to watch out for: If the subscriber handle is assigned to _ rather than a named binding, Rust drops it immediately at the end of the statement. Assign to a named variable or use .background().

Queryables and Storage

This chapter covers queryables — entities that reply to queries — and the storage pattern that combines a subscriber with a queryable to implement distributed key-value storage.

What Queryables Are

A queryable is an entity that registers to answer queries on a key expression. When session.get() is called, the routing fabric forwards the query to all matching queryables, which then send replies back to the caller. Queryables support four patterns: data-at-rest retrieval, remote computation, RPC-style invocation, and map-reduce aggregation.

What it is: The server side of zenoh’s query model. A queryable declares an interest in receiving queries and is responsible for generating one or more replies for each query it receives.

When to use it: Use a queryable whenever you want to expose data or computation that is retrieved on demand rather than continuously streamed. Unlike publishers, queryables respond only when asked.

What to watch out for: A queryable must call query.reply() or query.reply_err() for each received query. Failing to reply does not cause an error at the queryable, but the querier will wait until its timeout expires before closing the reply receiver.

Declaring a Queryable

session.declare_queryable(key_expr) registers the queryable with the routing fabric. Returns a Queryable handle backed by a channel. Incoming queries are consumed with recv_async().

How to use it:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let key_expr = "demo/example/zenoh-rs-queryable";
    let queryable = session
        .declare_queryable(key_expr)
        .await
        .unwrap();

    while let Ok(query) = queryable.recv_async().await {
        println!("Received query: {}", query.selector());
        query
            .reply(key_expr, "Queryable from Rust")
            .await
            .unwrap_or_else(|e| println!("Reply error: {e}"));
    }
}

What to watch out for: The queryable handle must remain in scope for the queryable to be active. Dropping it undeclares the queryable and releases routing resources. Assign it to a named variable, not _.

The Query Object

A Query carries the full context of the incoming request.

What it is: The incoming request object passed to a queryable, containing the key expression, optional parameters, and an optional request body.

When to use it: Inspect query fields to decide what data to return — filter by parameters, branch on the presence of a request body, or match against sub-keys.

How to use it:

The fields available on a Query:

• selector() — the full selector: key expression plus optional query string (e.g., demo/data/**?start=0)
• key_expr() — the key expression part of the selector only
• parameters() — the query string parameters after ? (e.g., start=0;end=100)
• payload() — optional request body for RPC-style use; None if the querier sent no body
• encoding() — the encoding of the request body

The queryable sends replies by calling query.reply(key, value). Each reply is a separate Sample delivered to the querier.

What to watch out for: query.reply() takes ownership of the reply value. If you need to send multiple replies from the same data, clone the value before each call.

Query with Body (RPC Pattern)

A query can carry a payload, enabling RPC-style invocation. The queryable receives the request body via query.payload() and uses it to compute a response.

What it is: A use of the query model where the payload acts as a function argument and the reply acts as a return value.

When to use it: Use this pattern for remote computation — image processing, inference, aggregation — where the input data is small enough to fit in a query payload and the output is returned as one or more replies.

How to use it:

#![allow(unused)]
fn main() {
while let Ok(query) = queryable.recv_async().await {
    match query.payload() {
        None => println!("No request body"),
        Some(body) => {
            let req = body.try_to_string().unwrap_or_else(|e| e.to_string().into());
            println!("Request: {req}");
        }
    }
    query.reply("demo/example/result", "computed result").await.unwrap_or_else(|e| println!("Error: {e}"));
}
}

What to watch out for: The reply key expression must match or be contained within the queryable’s declared key expression. Replying with a key that does not intersect the query key expression is not an error at the API level, but the router may drop it, and the querier may not receive it.

Completeness

complete(true) declares that this queryable holds the complete set of data for its key expression. No other queryable is needed to fully answer a query on this key space. The router uses this flag to optimize query routing — it can stop at the first complete match rather than forwarding to all matching queryables.

What it is: A hint to the routing fabric that this queryable is the authoritative source for its declared key expression.

When to use it: Set complete(true) only when your queryable genuinely holds all current state for its declared key space. The storage pattern (described below) is the primary use case.

How to use it:

#![allow(unused)]
fn main() {
let queryable = session
    .declare_queryable("demo/data/**")
    .complete(true)
    .await
    .unwrap();
}

What to watch out for: Setting complete(true) incorrectly causes queries to be routed only to your queryable even when other sources hold relevant data. Use complete(false) (the default) when your queryable handles a subset of queries or supplements other queryables.

Storage Pattern

A storage combines a subscriber (to receive and store incoming puts and deletes) with a queryable (to answer get requests from the stored data). This is the canonical pattern for implementing a distributed key-value store in an application.

What it is: A compound pattern that maintains an in-memory map of the latest sample per key and exposes that map to queriers.

When to use it: Use this pattern when you need application-level storage with custom logic, or when you want to store data in-process without running a separate zenohd instance with the storage plugin.

How to use it:

use std::collections::HashMap;
use futures::select;
use zenoh::{
    key_expr::{keyexpr, KeyExpr},
    sample::{Sample, SampleKind},
    Config,
};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let key_expr = "demo/example/**";
    let mut stored: HashMap<String, Sample> = HashMap::new();

    let subscriber = session.declare_subscriber(key_expr).await.unwrap();
    let queryable = session.declare_queryable(key_expr).complete(true).await.unwrap();

    loop {
        select! {
            sample = subscriber.recv_async() => {
                let sample = sample.unwrap();
                let key = sample.key_expr().to_string();
                match sample.kind() {
                    SampleKind::Put => { stored.insert(key, sample); }
                    SampleKind::Delete => { stored.remove(&key); }
                }
            },
            query = queryable.recv_async() => {
                let query = query.unwrap();
                for (stored_key, sample) in &stored {
                    if query.key_expr().intersects(unsafe {
                        keyexpr::from_str_unchecked(stored_key)
                    }) {
                        query.reply(sample.key_expr().clone(), sample.payload().clone())
                            .await
                            .unwrap_or_else(|e| println!("Reply error: {e}"));
                    }
                }
            }
        }
    }
}

This pattern requires the futures crate for the select! macro.

What to watch out for: The select! macro polls both futures concurrently on a single task. If reply processing is slow — for example, when the stored map is large — incoming subscriber samples queue up during that time. For high-throughput scenarios, consider spawning separate tasks for the subscriber and the queryable rather than interleaving them with select!.

Error Replies

Queryables can send error replies when processing fails. The querier receives these as Err(ReplyError) from reply.result().

What it is: A mechanism for queryables to signal failure to the querier without sending a valid sample.

When to use it: Send an error reply when the queryable cannot fulfill the request — for example, when the requested key does not exist, when required parameters are missing, or when computation fails.

How to use it:

#![allow(unused)]
fn main() {
query
    .reply_err("processing failed")
    .await
    .unwrap_or_else(|e| println!("Error: {e}"));
}

The querier receives this as reply.result() == Err(...) (see chapter 8).

What to watch out for: Sending an error reply does not prevent you from also sending successful replies. A queryable can send zero or more successful replies followed by an error reply, or vice versa. The querier receives all of them in the order they arrive.

zenohd Storage Plugin

For production storage, zenohd ships a storage plugin. Configure it in a JSON5 config file and start zenohd — no application-side code is required.

What it is: A router-side plugin that acts as a storage for a given key expression, answering queries from the data it has received via subscriptions.

When to use it: Use the zenohd storage plugin when you do not want to manage storage in your application process, when you need persistence across application restarts, or when the storage should be available independently of any particular application.

How to use it:

{
  plugins: {
    storage_manager: {
      storages: {
        demo: {
          key_expr: "demo/example/**",
          volume: { id: "memory" }
        }
      }
    }
  }
}

Start zenohd with this config: zenohd -c config.json5. The router then answers queries on demo/example/** without any application-side code.

What to watch out for: The memory volume does not persist data across zenohd restarts. For durable storage, configure a backend volume such as rocksdb or influxdb. Refer to the zenohd plugin documentation for the available backend options and their configuration keys.

Queries

This chapter covers the client side of the query model: issuing queries with session.get(), using declared queriers for repeated queries, query targeting, consolidation modes, and reply handling.

Issuing a Query

session.get(selector) sends a query to all matching queryables and storages. It returns a Receiver<Reply> that delivers replies asynchronously as they arrive from the network.

What it is: A one-shot query operation that broadcasts a request to all queryables whose key expression intersects the selector and collects their replies.

When to use it: Use session.get() for infrequent queries — startup data fetch, on-demand lookups, or administrative requests where the overhead of declaration is not justified.

How to use it:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    let replies = session.get("demo/example/**").await.unwrap();
    while let Ok(reply) = replies.recv_async().await {
        match reply.result() {
            Ok(sample) => {
                let payload = sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into());
                println!("'{}': '{}'", sample.key_expr().as_str(), payload);
            }
            Err(err) => {
                let payload = err.payload().try_to_string().unwrap_or_else(|e| e.to_string().into());
                println!("ERROR: '{payload}'");
            }
        }
    }
}

What to watch out for: The reply receiver closes after the query timeout expires (default 10 seconds). If you do not drain the receiver before the timeout, unconsumed replies are discarded. If your queryables are slow, set a longer timeout with .timeout(Duration).

Selectors

A selector is a key expression optionally followed by a query string: key/expr?param1=val1;param2=val2. The key expression part is used for routing. The parameters are delivered to queryables as query.parameters() and are interpreted entirely by the application.

What it is: The addressing mechanism for queries, extending key expressions with optional application-defined parameters.

When to use it: Use parameters for time-range queries, pagination, filtering hints, or any context that the queryable needs to construct the correct reply.

How to use it:

#![allow(unused)]
fn main() {
let replies = session
    .get("demo/data/**?start=0;end=100")
    .await
    .unwrap();
}

What to watch out for: Parameters are opaque to the zenoh routing layer. They are not parsed, validated, or used for routing — only the key expression part of the selector is used for that. Queryables receive the raw parameter string and must parse it themselves.

Query with Payload

Queries can carry a payload body for RPC-style use. The queryable receives the body via query.payload() and uses it to compute a response.

What it is: A query that carries input data in its body, enabling a request-reply interaction pattern analogous to a remote procedure call.

When to use it: Use a payload when you need to send input data to a remote computation — function arguments, image data, configuration parameters — and receive a computed result in return.

How to use it:

use std::time::Duration;
use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let replies = session
        .get("service/compute")
        .payload("input data")
        .timeout(Duration::from_millis(5000))
        .await
        .unwrap();
    while let Ok(reply) = replies.recv_async().await {
        match reply.result() {
            Ok(s) => println!("Result: {}", s.payload().try_to_string().unwrap_or_else(|e| e.to_string().into())),
            Err(e) => println!("Error: {}", e.payload().try_to_string().unwrap_or_else(|e| e.to_string().into())),
        }
    }
}

What to watch out for: Not all queryables inspect query.payload(). A queryable written to handle parameter-only queries will ignore the body. Ensure the queryable and querier agree on the interaction pattern before using payloads.

Query Targeting

target(QueryTarget::...) controls which queryables receive the query. The routing fabric uses the target to determine how broadly to forward the query.

What it is: A hint to the routing fabric about which subset of matching queryables should receive the query.

When to use it: Use All when you need replies from every matching source — for example, to aggregate sensor data from multiple nodes. Use BestMatching (the default) for normal lookups where one authoritative source is sufficient.

How to use it:

use zenoh::{query::QueryTarget, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let replies = session
        .get("demo/example/**")
        .target(QueryTarget::All)
        .await
        .unwrap();
    while let Ok(reply) = replies.recv_async().await {
        if let Ok(sample) = reply.result() {
            println!("{}: {}", sample.key_expr().as_str(), sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into()));
        }
    }
}

What to watch out for: QueryTarget::All can generate a large number of replies when many queryables match the selector. Ensure your application handles the full reply set, and consider using consolidation to reduce duplicates when querying multiple storages that hold overlapping data.

Timeout

Queries have a default timeout of 10 seconds. After the timeout, the reply receiver is closed and no further replies arrive, even if some queryables have not yet responded.

What it is: A deadline after which the query is considered complete and the reply receiver is sealed.

When to use it: Set an explicit timeout whenever the default 10-second window is too long (interactive UIs, latency-sensitive loops) or too short (slow storage backends, high-latency links).

How to use it:

#![allow(unused)]
fn main() {
use std::time::Duration;

let replies = session
    .get("demo/example/**")
    .timeout(Duration::from_millis(500))
    .await
    .unwrap();
}

What to watch out for: A timeout that is too short causes replies from slow queryables to be dropped silently. The querier receives no notification that replies were cut off — the receiver simply closes. If completeness matters, set a timeout that is generous relative to the worst-case queryable response time.

Consolidation

When multiple storages or queryables hold data for the same key, consolidation determines how duplicates are handled before they reach the application.

What it is: A policy for deduplicating replies that share the same key expression, using timestamps to determine which reply is the most recent.

When to use it: Use consolidation when querying multiple replicated storages to avoid processing the same logical value multiple times. Use None when you need every reply regardless of duplication, or when replies do not carry timestamps.

How to use it:

• ConsolidationMode::None — deliver all replies, including duplicates (default)
• ConsolidationMode::Monotonic — deduplicate using timestamps as replies arrive; keep the latest per key in a streaming fashion
• ConsolidationMode::Latest — buffer all replies, then deliver only the latest per key after the timeout

use zenoh::{query::ConsolidationMode, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let replies = session
        .get("demo/example/**")
        .consolidation(ConsolidationMode::Latest)
        .await
        .unwrap();
    while let Ok(reply) = replies.recv_async().await {
        if let Ok(sample) = reply.result() {
            println!("{}: {}", sample.key_expr().as_str(), sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into()));
        }
    }
}

What to watch out for: ConsolidationMode::Latest buffers all replies until the timeout expires, then delivers the deduplicated set. The application receives no replies until the full timeout has elapsed. This is not suitable for latency-sensitive use cases. Use Monotonic when you want deduplication with lower latency, accepting that a later reply might supersede one you already processed.

Declared Querier

session.declare_querier(key_expr) pre-declares a querier so the routing fabric can pre-establish paths to matching queryables. Use this for repeated queries to the same key expression to avoid per-query path resolution overhead.

What it is: A querier handle bound to a fixed key expression, with routing pre-negotiated at declaration time.

When to use it: Use a declared querier when issuing queries in a loop or at regular intervals to the same key expression. The per-query overhead is lower than repeated calls to session.get().

How to use it:

use std::time::Duration;
use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let querier = session
        .declare_querier("demo/example/**")
        .timeout(Duration::from_millis(5000))
        .await
        .unwrap();

    for _ in 0..10 {
        let replies = querier.get().await.unwrap();
        while let Ok(reply) = replies.recv_async().await {
            if let Ok(sample) = reply.result() {
                println!("{}: {}", sample.key_expr().as_str(), sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into()));
            }
        }
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
}

What to watch out for: The declared querier is bound to the key expression set at declaration time. To query a different key expression, declare a separate querier. As with all declared entities, the querier handle must remain in scope for the declaration to remain active.

Reply Types

Every Reply delivers its outcome through result().

What it is: The result type wrapping either a successful data sample or an error sent by the queryable.

When to use it: Always match on reply.result(). Ignoring error replies means your application silently drops failure signals from queryables.

How to use it:

• Ok(Sample) — a successful data reply; the sample carries the key, payload, encoding, and kind
• Err(ReplyError) — an error reply sent by the queryable via query.reply_err()
• A delete reply arrives as Ok(Sample) with sample.kind() == SampleKind::Delete

#![allow(unused)]
fn main() {
while let Ok(reply) = replies.recv_async().await {
    match reply.result() {
        Ok(sample) => {
            match sample.kind() {
                zenoh::sample::SampleKind::Put => {
                    println!("data: {}", sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into()));
                }
                zenoh::sample::SampleKind::Delete => {
                    println!("deleted: {}", sample.key_expr().as_str());
                }
            }
        }
        Err(err) => {
            println!("queryable error: {}", err.payload().try_to_string().unwrap_or_else(|e| e.to_string().into()));
        }
    }
}
}

What to watch out for: The Err variant carries a ReplyError which has a payload() method, not a standard Rust error message. Extract the error description by calling err.payload().try_to_string() rather than using the Display or Debug format of ReplyError directly, unless the error is a simple string.

Data Forwarding

The zenoh_ext crate provides a SubscriberForward extension that chains a subscriber directly to a publisher. This is the pattern used by z_forward.rs: every sample received on a source key expression is re-published on a destination key expression.

What it is: A utility that bridges two key expressions — receives from one, publishes to the other — without any application-level loop.

When to use it: Protocol bridging, namespace aliasing, and fan-out replication where no transformation is needed.

How to use it:

use zenoh::Config;
use zenoh_ext::SubscriberForward;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    let mut subscriber = session
        .declare_subscriber("demo/example/**")
        .await
        .unwrap();
    let publisher = session
        .declare_publisher("demo/forward")
        .await
        .unwrap();

    // Forward all samples from the subscriber to the publisher.
    subscriber.forward(publisher).await.unwrap();
}

Add zenoh-ext to Cargo.toml:

zenoh-ext = "1.4.0"

What to watch out for: forward() consumes the subscriber and blocks until the subscriber channel is closed. Run it in a dedicated task if you need to do other work concurrently.

Data Serialization

This chapter covers how Zenoh handles payload bytes: the ZBytes container, built-in conversions, the deserialize pattern, and encoding metadata.

ZBytes

ZBytes is Zenoh’s owned byte container. All payloads — puts, subscriber samples, queryable replies — carry a ZBytes. It stores a sequence of bytes without imposing a schema.

What it is

ZBytes wraps an internal sequence of byte slices. It owns its data and implements Send + Sync, making it safe to move across async task boundaries. It has no inherent schema or type information; that concern belongs to Encoding (covered below) and to the application.

When to use it

Every time you send or receive data in Zenoh, you interact with ZBytes. You create one when publishing, and you consume one when receiving a sample or reply.

How to use it

Creating ZBytes from common types via From:

#![allow(unused)]
fn main() {
use zenoh::bytes::ZBytes;

let b: ZBytes = "hello".into();
let b: ZBytes = 42i32.into();
let b: ZBytes = 3.14f64.into();
let b: ZBytes = vec![1u8, 2, 3].into();
}

Converting back:

#![allow(unused)]
fn main() {
// To string
let s = zbytes.try_to_string().unwrap_or_else(|e| e.to_string().into());

// To typed value via deserialize
let n: i32 = zbytes.deserialize().unwrap();

// To raw bytes
let v: Vec<u8> = zbytes.into();
}

What to watch out for

try_to_string returns a Cow<str>. If the bytes are valid UTF-8, it borrows from the internal buffer. If not, it returns an error, which you must handle explicitly. Do not use unwrap() blindly on untrusted data.

Primitives and ZBytes

ZBytes supports direct conversions for all primitive Rust types. The From and Into implementations cover bool, u8 through u128, i8 through i128, f32, and f64. The serialized format is the native little-endian representation of the type.

When to use it

Use primitive conversions when you control both the publisher and subscriber and both run Zenoh Rust. If interoperability with other languages or serialization formats is required, use a structured format (see the section on structured types below).

How to use it

Publisher side, following the z_put_float.rs pattern:

use zenoh::{bytes::Encoding, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    // Float
    let value: f64 = 3.14159;
    session
        .put("demo/sensors/temperature", value)
        .encoding(Encoding::APPLICATION_FLOAT64)
        .await
        .unwrap();

    // Integer
    let counter: i64 = 42;
    session
        .put("demo/counter", counter)
        .encoding(Encoding::APPLICATION_INTEGER)
        .await
        .unwrap();
}

Subscriber side, deserializing by type:

#![allow(unused)]
fn main() {
while let Ok(sample) = subscriber.recv_async().await {
    let temperature: f64 = sample.payload().deserialize().unwrap();
    println!("Temperature: {temperature}");
}
}

What to watch out for

The deserialize method is type-directed: it infers the target type from context. Ensure the type on the subscriber matches the type used by the publisher. Mismatched sizes (e.g., receiving as f32 what was sent as f64) produce a deserialization error, not a silent truncation.

Encoding

The Encoding struct conveys the format of the payload. It is metadata — Zenoh does not enforce it, but applications use it to select the right deserializer.

What it is

Encoding is a string-based identifier attached to a publication or reply. It is carried in the message header and available on received samples via sample.encoding(). It has no effect on routing or delivery; it is purely advisory.

When to use it

Set encoding whenever the payload format is not self-describing. This allows subscribers to inspect the encoding before attempting deserialization and to reject or convert formats they do not recognize.

How to use it

Standard constants:

Custom encodings for application-specific formats:

#![allow(unused)]
fn main() {
use zenoh::bytes::Encoding;

let enc = Encoding::new("application/protobuf");
session
    .put("demo/proto", serialized_bytes)
    .encoding(enc)
    .await
    .unwrap();
}

On the receive side:

#![allow(unused)]
fn main() {
while let Ok(sample) = subscriber.recv_async().await {
    let enc = sample.encoding();
    println!("Received with encoding: {enc}");
}
}

What to watch out for

Encoding is not enforced. A publisher can declare APPLICATION_JSON and send binary garbage; Zenoh will not reject it. Validation is the application’s responsibility. Treat the encoding field as a hint, not a guarantee, when receiving from untrusted publishers.

Iterating Bytes

ZBytes can hold a sequence of contiguous byte slices internally. Direct iteration over those slices avoids a copy when you only need to read the bytes.

When to use it

Use slice iteration when passing the raw bytes directly to a parser (e.g., a protobuf decoder) that accepts &[u8] and you want to avoid a heap allocation. For most cases, collecting to Vec<u8> is simpler and sufficient.

How to use it

Iterate the raw slices:

#![allow(unused)]
fn main() {
for slice in zbytes.slices() {
    // process raw &[u8]
}
}

Collect to Vec<u8>:

#![allow(unused)]
fn main() {
let bytes: Vec<u8> = zbytes.into();
}

What to watch out for

The slices() iterator yields &[u8] references that borrow from ZBytes. The ZBytes value must remain alive for the duration of the iteration. Moving or dropping it while iterating is a compile error. If you need to send the bytes across an await point, collect them first.

Structured Types with z_formats

Zenoh provides a z_serialize/z_deserialize pattern for types that implement Serialize/Deserialize from serde. For JSON payloads, serde_json is the standard choice. This maps to the z_formats.rs example.

What it is

Rather than manually converting to and from Vec<u8>, you serialize your domain types directly to a byte representation and attach the appropriate encoding. The subscriber reverses the process.

When to use it

Use structured serialization when your payloads carry typed, multi-field data: sensor readings, commands, status messages. Prefer JSON for interoperability with non-Rust systems. Prefer a compact binary format (protobuf, CBOR, MessagePack) for bandwidth-constrained links.

How to use it

Add dependencies to Cargo.toml:

[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
zenoh = "1"
tokio = { version = "1", features = ["full"] }

Publisher:

use serde::{Deserialize, Serialize};
use zenoh::Config;

#[derive(Serialize, Deserialize, Debug)]
struct SensorReading {
    temperature: f64,
    humidity: f64,
}

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    let reading = SensorReading {
        temperature: 23.5,
        humidity: 60.0,
    };
    let payload = serde_json::to_vec(&reading).unwrap();

    session
        .put("demo/sensor/reading", payload)
        .encoding(zenoh::bytes::Encoding::APPLICATION_JSON)
        .await
        .unwrap();
}

Subscriber side:

#![allow(unused)]
fn main() {
while let Ok(sample) = subscriber.recv_async().await {
    let bytes: Vec<u8> = sample.payload().into();
    let reading: SensorReading = serde_json::from_slice(&bytes).unwrap();
    println!("{reading:?}");
}
}

What to watch out for

serde_json::to_vec allocates. For high-frequency small messages, consider pre-allocating a buffer with serde_json::to_writer targeting a Vec<u8> that you reuse across iterations. Also consider that unwrap() on deserialization will panic if the remote publisher sends a malformed payload; use match or ? in production code.

Attachments

Samples and queries can carry an attachment: a secondary ZBytes for metadata. Attachments are separate from the main payload and are not interpreted by the routing layer. Common uses are correlation IDs, distributed trace context, and sequence numbers.

What it is

An attachment is an optional ZBytes carried alongside the primary payload. It is available on received samples via sample.attachment() and on received queries via query.attachment().

When to use it

Use attachments when you need to carry out-of-band metadata without modifying the primary payload schema. This is preferable to embedding metadata fields in the payload itself when the payload schema is defined externally (e.g., a protobuf schema you do not control).

How to use it

Publishing with an attachment:

#![allow(unused)]
fn main() {
publisher
    .put("data")
    .attachment("trace-id-1234")
    .await
    .unwrap();
}

Receiving the attachment:

#![allow(unused)]
fn main() {
if let Some(att) = sample.attachment() {
    let id = att.try_to_string().unwrap_or_else(|e| e.to_string().into());
    println!("attachment: {id}");
}
}

Attachments on queries (from the queryable side):

#![allow(unused)]
fn main() {
while let Ok(query) = queryable.recv_async().await {
    if let Some(att) = query.attachment() {
        let meta = att.try_to_string().unwrap_or_else(|e| e.to_string().into());
        println!("query attachment: {meta}");
    }
    query.reply(query.key_expr(), "response").await.unwrap();
}
}

What to watch out for

Attachments are not indexed or filtered by the router. You cannot route or filter by attachment content. If you need to route on metadata, encode it in the key expression instead.

Quality of Service

This chapter covers Zenoh’s QoS knobs: reliability, priority, congestion control, and express mode. These are set per-publisher or per-put.

Overview

Zenoh QoS controls four independent axes:

1. Reliability — should delivery be acknowledged?
2. Priority — what traffic class?
3. Congestion control — what happens when buffers are full?
4. Express — should batching be bypassed?

These can be set on declare_publisher(). Reliability, priority, and congestion control can also be overridden per-put on a declared publisher. Express is a publisher-level setting only.

The defaults are designed for high-throughput telemetry: best-effort, normal priority, drop on congestion, with batching enabled. Override them when your data has different requirements.

Reliability

What it is

Reliability controls whether Zenoh retransmits messages that are lost in transit.

• Reliability::BestEffort (default) — no retransmission. The message is sent once; if a packet is dropped, the receiver never learns of it.
• Reliability::Reliable — acknowledged delivery. Lost messages are retransmitted until acknowledged or the session is torn down.

When to use it

Use BestEffort for high-frequency telemetry where a missed sample is acceptable and low overhead matters: sensor streams, video feeds, log tails.

Use Reliable for commands, configuration changes, software deployments, and any data where loss causes incorrect behavior.

How to use it

use zenoh::{qos::Reliability, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let publisher = session
        .declare_publisher("control/cmd")
        .reliability(Reliability::Reliable)
        .await
        .unwrap();
    publisher.put("engage").await.unwrap();
}

What to watch out for

Reliability is negotiated end-to-end. If any hop on the path uses a transport that does not support reliability (e.g., UDP with BestEffort), the guarantee degrades to BestEffort on that segment. Reliable delivery over a UDP-only path is not possible; use TCP or TLS transports for reliable sessions. Verify your transport configuration when Reliable is required.

Priority

What it is

Priority assigns a traffic class to messages. Seven levels exist, in decreasing order of priority:

1. Priority::RealTime
2. Priority::InteractiveHigh
3. Priority::InteractiveLow
4. Priority::DataHigh
5. Priority::Data (default)
6. Priority::DataLow
7. Priority::Background

Higher-priority messages are scheduled before lower-priority ones in the router queues. Priority is respected end-to-end: every Zenoh router on the path schedules by priority class.

When to use it

Use priority in mixed-criticality systems where control messages must not be delayed by bulk data transfers. For example, a robot controller that publishes both safety-critical brake commands and low-priority telemetry benefits from assigning RealTime to the brake publisher and Background to the telemetry publisher.

How to use it

use zenoh::{qos::Priority, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    let realtime = session
        .declare_publisher("actuator/brake")
        .priority(Priority::RealTime)
        .await
        .unwrap();

    let telemetry = session
        .declare_publisher("telemetry/gps")
        .priority(Priority::DataLow)
        .await
        .unwrap();

    realtime.put("brake_now").await.unwrap();
    telemetry.put("48.8566, 2.3522").await.unwrap();
}

What to watch out for

Priority scheduling only takes effect when there is contention. On an idle network, all priorities are equivalent. Priority does not guarantee a latency bound; it only influences ordering relative to other messages in the queue. Do not use priority as a substitute for network capacity planning.

Congestion Control

What it is

Congestion control determines what Zenoh does when the output buffer is full and a new message arrives.

• CongestionControl::Drop (default) — the new message is discarded. The sender is not blocked.
• CongestionControl::Block — the send call blocks until buffer space becomes available.

When to use it

Use Drop for data that can be replaced by the next sample: video frames, sensor readings, position updates. A dropped frame is acceptable if the next one arrives promptly.

Use Block for data where every message must be delivered: financial transactions, command sequences, firmware chunks. Dropping any message is a correctness error.

How to use it

use zenoh::{qos::CongestionControl, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let publisher = session
        .declare_publisher("stream/video")
        .congestion_control(CongestionControl::Drop)
        .await
        .unwrap();
    publisher.put(vec![0u8; 1024 * 100]).await.unwrap();
}

What to watch out for

CongestionControl::Block can cause the publisher to stall indefinitely if the consumer is stopped or the network link is saturated. Only use Block with Reliability::Reliable and in situations where you can tolerate unbounded producer latency. In async code, a blocked send occupies the executor task until the buffer drains. If all executor threads are stalled on blocked sends, the application deadlocks. Consider using a timeout around blocked publishers in production code.

Express Mode

What it is

By default, Zenoh batches small messages to amortize per-packet overhead and improve throughput. Express mode disables this batching and sends each message immediately after it is queued.

When to use it

Use express(true) when latency matters more than throughput: IMU data at 1 kHz, control signals in a feedback loop, RTT-sensitive request-response protocols. The throughput penalty relative to batched mode is roughly 10-30% for small messages, depending on message size and network conditions.

How to use it

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let publisher = session
        .declare_publisher("sensor/imu")
        .express(true)
        .await
        .unwrap();
    publisher.put(vec![0u8; 48]).await.unwrap();
}

What to watch out for

Express mode trades throughput for latency. Enabling it on a high-volume stream that is not latency-sensitive increases CPU and network overhead without benefit. Measure the actual latency improvement in your environment before enabling it unconditionally. On loopback or gigabit LAN, the difference may be negligible.

Setting QoS on Individual Puts

What it is

A declared publisher carries default QoS settings. These defaults can be overridden on a per-put basis by chaining QoS methods on the put builder. The override applies only to that one publication; subsequent puts revert to the publisher defaults.

When to use it

Use per-put overrides when a single publisher emits messages of varying criticality. For example, a publisher that normally sends background telemetry might occasionally emit a high-priority alert.

How to use it

use zenoh::{qos::Priority, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let publisher = session
        .declare_publisher("demo/data")
        .await
        .unwrap();

    // High-priority put
    publisher
        .put("critical")
        .priority(Priority::RealTime)
        .await
        .unwrap();

    // Background put using publisher defaults
    publisher
        .put("background data")
        .priority(Priority::Background)
        .await
        .unwrap();
}

What to watch out for

Per-put QoS does not override the publisher’s reliability setting; that is fixed at declaration time. Only priority, congestion control, and express can be set per-put. If you need to send the same key expression with different reliability levels, declare two separate publishers.

QoS Summary Table

Routing and Topology

This chapter explains how Zenoh sessions discover each other and how data flows across the network in peer, client, and router modes.

Three Modes

Zenoh has three session modes, set via the mode config field:

A session declares its mode at open time. The mode cannot be changed after the session is open.

Peer Mode

What it is

Peer mode is the default. Each peer participates in discovery and routes data directly to other reachable peers. No central infrastructure is required. Two peers on the same L2 segment can exchange data without any router.

When to use it

Peer mode is suitable for LAN environments, robotics labs, and development. Use it when all participants are on the same network segment or can reach each other via TCP without NAT.

How to use it

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    // Config::default() uses peer mode with multicast scouting.
    let session = zenoh::open(Config::default()).await.unwrap();
    println!("Running as peer");
    std::future::pending::<()>().await;
}

What to watch out for

In peer mode, every peer maintains its own routing table and forwards relevant traffic to reachable peers. As the number of peers grows, the overhead of maintaining pairwise connections increases. For deployments with hundreds of nodes, consider client mode with dedicated routers to centralize the routing burden.

Client Mode

What it is

Client mode is a thin mode: the client connects to one or more routers and delegates all routing to them. The client itself does not route any data. From the router’s perspective, a client is a leaf node.

When to use it

Use client mode for constrained nodes with limited CPU or memory, mobile nodes that move between networks, and nodes behind NAT where peer-to-peer connectivity is not possible. A client needs only a TCP connection to a known router address.

How to use it

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let mut config = Config::default();
    config.insert_json5("mode", r#""client""#).unwrap();
    config
        .insert_json5("connect/endpoints", r#"["tcp/192.168.1.10:7447"]"#)
        .unwrap();
    let session = zenoh::open(config).await.unwrap();
    println!("Running as client");
    std::future::pending::<()>().await;
}

What to watch out for

If all configured routers are unreachable, the client session fails to open. In production, configure multiple router endpoints for redundancy. Client sessions do not re-route through alternative peers if the router goes down; the session must be re-opened with a different endpoint.

Router Mode and zenohd

What it is

zenohd is the reference router implementation. It runs as a daemon and provides full routing table management, subscription aggregation, and inter-router connectivity. Routers form a mesh: each connects to others and shares routing tables. They route data between clients and peers that cannot reach each other directly.

When to use it

Deploy routers as infrastructure nodes at network boundaries, in datacenters, and as entry points for WAN-connected clients. A router is required when clients on different L2 segments need to exchange data.

How to use it

Run the router with default settings:

zenohd

Run with a config file:

zenohd -c /etc/zenoh/config.json5

Key config fields for a router:

{
  mode: "router",
  listen: {
    endpoints: ["tcp/0.0.0.0:7447"]
  },
  connect: {
    endpoints: ["tcp/other-router.example.com:7447"]
  }
}

What to watch out for

A router that connects to no other routers is an island. Clients connected to it can only communicate with each other. For inter-region connectivity, configure explicit connect/endpoints to peer routers. Monitor router memory usage: a router holds subscription state for all connected clients and peers.

Multicast Scouting

What it is

In peer mode, Zenoh sends scout messages on 224.0.0.224:7446 (UDP multicast). All Zenoh peers on the same L2 segment respond with their locators, enabling automatic discovery without any prior configuration.

When to use it

Multicast scouting is appropriate for development environments, lab networks, and any deployment where all participants share an L2 segment and multicast is permitted. It requires no configuration: Config::default() enables it.

How to use it

Multicast scouting is active by default. To disable it for production environments or firewalled networks:

#![allow(unused)]
fn main() {
let mut config = Config::default();
config
    .insert_json5("scouting/multicast/enabled", "false")
    .unwrap();
}

What to watch out for

Many cloud and datacenter networks block multicast. If peers fail to discover each other, verify that 224.0.0.224 is reachable on the local segment. On Docker or Kubernetes networks, multicast is typically unavailable; use static peer configuration instead (see the section below).

Gossip Scouting

What it is

When multicast is unavailable, Zenoh routers gossip peer information to connected clients. A client connecting to a router learns about other peers indirectly through the router’s advertisement of known locators. Gossip scouting is an alternative discovery mechanism for networks where multicast is blocked.

When to use it

Enable gossip scouting in production environments where multicast is not available but you still want dynamic peer discovery mediated by routers.

How to use it

Router-side configuration:

{
  scouting: {
    gossip: {
      enabled: true
    }
  }
}

What to watch out for

Gossip discovery adds latency relative to multicast: a client must first connect to a router before learning about other peers. In time-sensitive startup scenarios, supplement gossip with static peer endpoints so that critical connections are established before gossip converges.

Static Peers

What it is

Static peer configuration bypasses discovery entirely. The session connects directly to the listed endpoints at startup. No scout messages are sent or processed for those endpoints.

When to use it

Use static configuration in datacenter and production environments where addresses are known, stable, and multicast is unavailable. Static configuration is also appropriate when you need deterministic startup behavior without waiting for scouting to converge.

How to use it

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let mut config = Config::default();
    config
        .insert_json5("scouting/multicast/enabled", "false")
        .unwrap();
    config
        .insert_json5(
            "connect/endpoints",
            r#"["tcp/10.0.0.1:7447", "tcp/10.0.0.2:7447"]"#,
        )
        .unwrap();
    let session = zenoh::open(config).await.unwrap();
    std::future::pending::<()>().await;
}

What to watch out for

If none of the configured endpoints are reachable, the session opens in a degraded state or fails, depending on configuration. For peer mode with static endpoints, the session proceeds even if some endpoints are down, but those peers are not reachable. Test your failover behavior explicitly.

z_scout: Discovering the Network

What it is

zenoh::scout() sends scouting messages and reports discovered peers and routers without opening a full session. It returns an asynchronous receiver of Hello messages.

When to use it

Use scout for network diagnostics, topology inspection, and tooling that needs to enumerate the network without joining it as a participant.

How to use it

use zenoh::{config::WhatAmI, scout, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let receiver = scout(WhatAmI::Peer | WhatAmI::Router, Config::default())
        .await
        .unwrap();

    let _ = tokio::time::timeout(std::time::Duration::from_secs(1), async {
        while let Ok(hello) = receiver.recv_async().await {
            println!("{hello}");
        }
    })
    .await;

    receiver.stop();
}

What to watch out for

scout uses the same multicast address as normal scouting. In environments where multicast is disabled, it discovers nothing. The timeout wrapper is important: without it, recv_async blocks indefinitely waiting for responses that will not arrive. Always call receiver.stop() when done to release the multicast socket.

Regionalization

What it is

Large deployments partition the network into regions. Each region has local routers that handle intra-region traffic. Inter-region traffic flows through designated backbone routers that connect the regions. This limits the blast radius of routing updates and subscription state to the relevant region.

When to use it

Use regionalization in deployments with hundreds or thousands of nodes, particularly when nodes in different geographic locations or administrative domains should be isolated from each other’s traffic by default.

How to use it

Key expressions carry the region scope by convention. A common pattern is to prefix all key expressions with a region identifier:

region/europe/sensors/temperature
region/asia/sensors/temperature

Backbone routers are configured to forward only subscriptions that match inter-region patterns. Intra-region traffic stays within the region’s router mesh. The routing fabric forwards only relevant subscriptions across region boundaries based on declared interests.

What to watch out for

Regionalization is a configuration and naming discipline, not an automatic feature. Misconfigured backbone router routing rules can leak intra-region traffic across regions or silently black-hole inter-region subscriptions. Test cross-region delivery explicitly in your staging environment.

Firewall Considerations

Zenoh uses the following ports by default. Ensure they are open on any firewall between communicating nodes:

• Multicast UDP scouting: 224.0.0.224:7446
• Zenoh TCP transport: port 7447
• Zenoh TLS transport: port 7448

For routers on public IP addresses, use explicit endpoint configuration and disable multicast:

{
  mode: "router",
  listen: {
    endpoints: ["tls/0.0.0.0:7448"]
  },
  scouting: {
    multicast: {
      enabled: false
    }
  }
}

TLS is recommended for any router exposed on a public network. See the TLS configuration documentation for certificate setup.

Liveliness

This chapter covers Zenoh’s liveliness protocol: declaring tokens to signal presence, subscribing to presence changes, and querying for currently alive tokens.

What Liveliness Is

A liveliness token is a resource that exists as long as the session that declared it is alive. When the session closes — gracefully or due to failure — the token disappears and all liveliness subscribers receive a Delete sample for that token’s key expression. This enables lightweight service registry, peer heartbeat, and presence tracking without explicit heartbeat messages.

The liveliness protocol is built into Zenoh’s session layer. It does not require application-level timers or polling. The router detects session loss through transport-level keepalives and propagates the deletion event automatically.

Declaring a Liveliness Token

What it is

session.liveliness().declare_token(key_expr) declares a token under the given key expression. From the moment the token is declared, any liveliness subscriber matching that key expression receives a Put sample. The token remains active until it is explicitly undeclared or until the session that declared it closes.

When to use it

Declare a liveliness token when your application component wants to advertise its presence on the network. The key expression should encode enough identity information for consumers to act on: service type, instance ID, region, or any other relevant attributes.

How to use it

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let token = session
        .liveliness()
        .declare_token("group1/zenoh-rs")
        .await
        .unwrap();

    println!("Liveliness token declared. Press Ctrl-C to quit.");
    std::future::pending::<()>().await;

    // Token is automatically undeclared when dropped.
    // Explicit undeclaration:
    // token.undeclare().await.unwrap();
    drop(token);
}

What to watch out for

A token is undeclared when the LivelinessToken value is dropped, even if no explicit undeclare() call is made. This is intentional: it ensures cleanup on panic or early return. However, it also means that storing the token in a local variable that goes out of scope before the application ends will silently undeclare it. Store the token in a variable whose lifetime matches the intended presence duration — typically the main function or a long-lived struct field.

Subscribing to Liveliness Changes

What it is

session.liveliness().declare_subscriber(key_expr) creates a subscriber that receives Put events when a matching token is declared and Delete events when a matching token disappears. With .history(true), the subscriber also receives Put events for tokens that were already alive at the time the subscriber was declared.

When to use it

Subscribe to liveliness changes when your application needs to react to the appearance or disappearance of other components: updating a service registry, triggering reconnection logic, alerting on unexpected departures, or building a live dashboard of active nodes.

How to use it

use zenoh::{sample::SampleKind, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let subscriber = session
        .liveliness()
        .declare_subscriber("group1/**")
        .history(true)
        .await
        .unwrap();

    while let Ok(sample) = subscriber.recv_async().await {
        match sample.kind() {
            SampleKind::Put => println!("Alive: {}", sample.key_expr().as_str()),
            SampleKind::Delete => println!("Gone: {}", sample.key_expr().as_str()),
        }
    }
}

What to watch out for

Without .history(true), a subscriber only learns about tokens declared after it was created. If the subscriber starts after some tokens are already alive, it misses their initial Put events and has no knowledge of them until they disappear or reappear. Always use .history(true) when building a complete view of current presence. With .history(true), the initial burst of Put events for pre-existing tokens arrives before any subsequent change events, but you should not assume any particular ordering within that burst.

Querying Current Live Tokens

What it is

session.liveliness().get(key_expr) is a one-shot query that retrieves all currently active tokens matching the key expression. It does not subscribe to future changes; it captures a point-in-time snapshot of liveness state.

When to use it

Use get at startup to enumerate existing live tokens before subscribing to changes, or in diagnostic tools that need to inspect current presence without maintaining a persistent subscription. Combine get with a subscriber to build a complete, consistent view: first get the current state, then subscribe to deltas.

How to use it

use std::time::Duration;
use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let replies = session
        .liveliness()
        .get("group1/**")
        .timeout(Duration::from_millis(5000))
        .await
        .unwrap();

    while let Ok(reply) = replies.recv_async().await {
        match reply.result() {
            Ok(sample) => println!("Alive: {}", sample.key_expr().as_str()),
            Err(e) => println!(
                "Error: {}",
                e.payload()
                    .try_to_string()
                    .unwrap_or_else(|e| e.to_string().into())
            ),
        }
    }
}

What to watch out for

The get operation has a timeout. Tokens declared by sessions that are not reachable within the timeout window do not appear in the results. In large distributed deployments where routers add latency, set a generous timeout. A too-short timeout produces an incomplete snapshot that may cause the application to believe fewer services are available than actually are.

There is a race between get and declare_subscriber: tokens declared in the gap between the two calls may be missed. The correct pattern is to declare the subscriber first with .history(true) and then issue a get to populate initial state. The subscriber’s history replay and the get results may overlap; deduplicate by key expression.

Scalability Compared to DDS Liveliness

DDS liveliness requires O(n^2) heartbeats in a fully connected system: every participant must heartbeat every other participant to maintain liveness assertions. In a 1000-node system, this generates roughly 1,000,000 heartbeat messages per heartbeat period under the AUTOMATIC_LIVELINESS_QOS_POLICY.

Zenoh’s liveliness protocol is O(n): each token propagates through the routing fabric once as a single event. The router detects session loss through existing transport-level keepalives and generates a single delete event per lost token. In a 1000-node system, Zenoh generates 1000 presence events per heartbeat period — three orders of magnitude fewer messages.

This matters in practice for systems that add and remove nodes frequently, such as container orchestration environments, mobile robots, and vehicular networks. The DDS approach can consume a significant fraction of available bandwidth on the monitoring channel alone; the Zenoh approach does not.

Use Cases

Service Registry

Services declare tokens on services/<type>/<id>. Clients subscribe with services/<type>/** to discover available instances of a given service type. When a service goes down, clients receive a Delete event and can remove it from their routing table.

Fleet Tracking

Vehicles declare tokens on fleet/<vehicle-id>. A monitoring service subscribes to fleet/** with .history(true) to maintain a live roster of active vehicles. Unexpected Delete events trigger alerts.

Peer Health Monitoring

Each node declares a liveliness token on nodes/<node-id>. A watchdog subscribes to nodes/** and maintains a set of expected nodes. A Delete event for a node that should be running triggers a remediation workflow: restart, alert, or failover.

Leader Election

All candidates declare tokens on election/<cluster-id>/**. Candidates subscribe to the same prefix. The candidate whose token key expression sorts first (or by any deterministic rule applied to the key) is the leader. When the leader’s session closes, all remaining candidates receive a Delete event and re-evaluate leadership based on the surviving tokens.

Integration with Queryables

Liveliness tokens and queryables compose naturally. A service can declare both a liveliness token and a queryable on the same key expression:

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    // Advertise presence
    let _token = session
        .liveliness()
        .declare_token("services/temperature/sensor-1")
        .await
        .unwrap();

    // Serve queries
    let queryable = session
        .declare_queryable("services/temperature/sensor-1")
        .await
        .unwrap();

    while let Ok(query) = queryable.recv_async().await {
        query
            .reply(query.key_expr(), "23.5")
            .await
            .unwrap();
    }
}

Clients discover the service via the liveliness token and issue queries to the same key expression. When the service goes down, the liveliness token disappears and clients know not to issue further queries. This pattern avoids the need for a separate service discovery mechanism.

Shared Memory

This chapter covers Zenoh’s shared memory (SHM) transport: when to use it, how to allocate SHM buffers, zero-copy publishing, and receiving SHM data.

When SHM Applies

SHM transport is available only between processes on the same host. It provides zero-copy delivery: the publisher writes into a shared memory region, the subscriber reads directly from it without any data copy. This is the primary mechanism for high-throughput, low-latency intra-host communication in use cases such as camera frames, LiDAR point clouds, video streams, and ML inference data.

SHM is not a separate protocol — it is a transport optimization. The same publisher and subscriber API is used; Zenoh selects SHM automatically when available and configured.

The shared-memory feature must be enabled in Cargo.toml:

zenoh = { version = "1.4.0", features = ["shared-memory"] }

When the publisher and subscriber are on different hosts, Zenoh automatically falls back to TCP or QUIC transport. No code change is required on either side.

Creating an SHM Provider

The SHM provider manages the shared memory region. The default backend uses POSIX shared memory. ShmProviderBuilder::default_backend is the simplest way to create one.

use zenoh::{
    shm::{BlockOn, GarbageCollect, ShmProviderBuilder},
    Config, Wait,
};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    // Create a 1 MB SHM provider
    let provider = ShmProviderBuilder::default_backend(1024 * 1024)
        .wait()
        .unwrap();

    let publisher = session
        .declare_publisher("demo/shm/data")
        .await
        .unwrap();

    // Allocate a 256-byte buffer
    let mut sbuf = provider
        .alloc(256)
        .with_policy::<BlockOn<GarbageCollect>>()
        .await
        .unwrap();

    sbuf[..5].copy_from_slice(b"hello");

    // Zero-copy publish
    publisher.put(sbuf).await.unwrap();
}

BlockOn<GarbageCollect> means: if no buffer is immediately available, first run garbage collection to reclaim released buffers, then block if the pool is still exhausted. This is the recommended allocation policy for production use.

Allocation Policies

Use GarbageCollect without blocking only in real-time contexts where blocking is prohibited.

Receiving SHM Data

The subscriber API is identical to the non-SHM case. Zenoh handles the SHM-to-bytes mapping transparently.

use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    let subscriber = session
        .declare_subscriber("demo/shm/data")
        .await
        .unwrap();

    while let Ok(sample) = subscriber.recv_async().await {
        // payload may reference SHM directly — no copy occurs
        let bytes: Vec<u8> = sample.payload().into();
        println!("Received {} bytes", bytes.len());
    }
}

Avoid converting the payload to Vec<u8> in hot paths if the goal is zero-copy. Access the payload bytes directly through the ZBytes API to avoid materializing a copy.

SHM Queries and Queryables

SHM works transparently with queryables. Allocate an SHM buffer inside a queryable handler and return it as the reply payload.

#![allow(unused)]
fn main() {
// Inside a queryable handler:
let mut sbuf = provider
    .alloc(1024)
    .with_policy::<BlockOn<GarbageCollect>>()
    .await
    .unwrap();
// ... fill sbuf with reply data ...
query.reply("demo/shm/data", sbuf).await.unwrap_or_else(|e| println!("Error: {e}"));
}

The querier receives the reply with the same subscriber-side transparency: no code change is needed to handle an SHM-backed reply versus a TCP-backed reply.

Custom POSIX SHM Provider

For control over the POSIX region name and size, use PosixShmProviderBackend directly. This is useful when multiple providers must coexist in the same process with distinct memory regions.

#![allow(unused)]
fn main() {
use zenoh::{
    shm::{
        BlockOn, GarbageCollect, PosixShmProviderBackend, ShmProviderBuilder, POSIX_PROTOCOL_ID,
    },
    Wait,
};

let backend = PosixShmProviderBackend::builder()
    .with_size(4 * 1024 * 1024) // 4 MB
    .res()
    .wait()
    .unwrap();

let provider = ShmProviderBuilder::builder()
    .protocol_id::<POSIX_PROTOCOL_ID>()
    .backend(backend)
    .res()
    .await
    .unwrap();
}

Use this pattern when the default single-region provider is insufficient, or when you need to name the POSIX region explicitly for external tooling.

GPU and Accelerator Backends

Zenoh’s SHM provider interface is pluggable. Custom backends can wrap GPU memory (CUDA, ROCm), DMA buffers, or FPGA memory regions. The publisher-side API remains unchanged — only the provider construction differs.

This pattern is common in ML inference pipelines where the inference result lives in GPU memory and must be consumed by a downstream subscriber without a GPU-to-CPU copy. The custom backend presents the GPU memory as an SHM region; Zenoh routes the buffer reference rather than copying bytes.

Implementing a custom backend requires implementing the ShmProviderBackend trait from zenoh::shm.

Buffer Lifecycle

SHM buffers are reference-counted. The buffer is released back to the provider’s pool when all references are dropped, including any copy held by a subscriber. A subscriber that holds an SHM buffer reference longer than necessary depletes the provider pool, causing subsequent alloc() calls to block or fail.

Guidelines:

• Process or copy the payload in the subscriber and drop the sample promptly.
• Size the provider pool generously relative to the number of in-flight messages.
• Use ZENOH_TRACE_SHM=1 with ZENOH_LOG=trace to observe allocation and release events during debugging.

Security

This chapter covers Zenoh’s security features: TLS transport, mutual authentication, access control lists, and zero-trust configuration.

TLS Transport

Zenoh supports TLS 1.3 on the tls:// transport endpoint. TLS encrypts all data in transit and authenticates the server to the client. Use TLS whenever Zenoh nodes communicate over an untrusted network.

Configure a router to listen with TLS:

// zenohd config with TLS
{
  mode: "router",
  listen: {
    endpoints: ["tls/0.0.0.0:7448"]
  },
  transport: {
    unicast: {
      tls: {
        server_certificate: "/etc/zenoh/server.pem",
        server_private_key: "/etc/zenoh/server_key.pem",
        root_ca_certificate: "/etc/zenoh/ca.pem"
      }
    }
  }
}

Client connecting with TLS:

{
  mode: "client",
  connect: {
    endpoints: ["tls/router.example.com:7448"]
  },
  transport: {
    unicast: {
      tls: {
        root_ca_certificate: "/etc/zenoh/ca.pem"
      }
    }
  }
}

The client verifies the server certificate against root_ca_certificate. If verification fails, the connection is rejected.

Mutual Authentication (mTLS)

Mutual TLS requires both the server and the client to present certificates. The server verifies the client certificate, enabling identity-based access decisions at the transport layer.

Server config — require client certificate:

{
  transport: {
    unicast: {
      tls: {
        server_certificate: "/etc/zenoh/server.pem",
        server_private_key: "/etc/zenoh/server_key.pem",
        root_ca_certificate: "/etc/zenoh/ca.pem",
        client_auth: true
      }
    }
  }
}

Client config — provide certificate:

{
  transport: {
    unicast: {
      tls: {
        client_certificate: "/etc/zenoh/client.pem",
        client_private_key: "/etc/zenoh/client_key.pem",
        root_ca_certificate: "/etc/zenoh/ca.pem"
      }
    }
  }
}

With client_auth: true, the router rejects any client that does not present a valid certificate signed by the configured CA. The certificate’s Common Name (CN) is available to the ACL subsystem for per-identity rules.

Access Control Lists

Zenoh supports per-certificate, per-interface, and per-key-expression access control. Configure ACLs in the router JSON5 config.

{
  access_control: {
    enabled: true,
    default_permission: "deny",
    rules: [
      {
        id: "sensor-pub",
        messages: ["put", "delete"],
        key_exprs: ["sensors/**"],
        flows: ["egress"],
        permission: "allow"
      },
      {
        id: "sensor-sub",
        messages: ["declare_subscriber"],
        key_exprs: ["sensors/**"],
        flows: ["ingress"],
        permission: "allow"
      }
    ],
    subjects: [
      {
        id: "sensor-node",
        interfaces: ["eth0"],
        cert_common_names: ["sensor-device-01"]
      }
    ],
    policies: [
      {
        subjects: ["sensor-node"],
        rules: ["sensor-pub", "sensor-sub"]
      }
    ]
  }
}

With default_permission: "deny", only operations explicitly listed in an "allow" rule proceed. This is the zero-trust posture: deny everything, permit only what is explicitly needed.

ACL Fields

Token-Based Authentication

Zenoh supports username/password authentication in environments where certificate management is not feasible.

{
  transport: {
    unicast: {
      auth: {
        usrpwd: {
          user: "zenoh-client",
          password: "s3cr3t",
          dictionary_file: "/etc/zenoh/users.json5"
        }
      }
    }
  }
}

The dictionary_file maps usernames to hashed passwords. Token-based auth is a lower-security option compared to mTLS: it does not provide per-message integrity or forward secrecy unless combined with TLS.

Zero-Trust Summary

A zero-trust Zenoh deployment combines all of the above:

1. All transports use TLS with mutual authentication (client_auth: true).
2. Access control is enabled with default_permission: "deny".
3. Explicit allowlist rules are defined per certificate identity and key expression.
4. No unauthenticated peer connections are permitted.
5. Admin space write access is disabled or protected by ACL rules.

This configuration ensures that a compromised node cannot publish to unauthorized topics, subscribe to data it does not own, or modify router configuration at runtime.

Performance and Benchmarking

This chapter covers measuring Zenoh latency and throughput using the built-in benchmark examples, and tuning the key parameters that affect performance.

Built-in Benchmark Tools

The Zenoh repository ships four benchmark examples:

• z_ping / z_pong — round-trip latency measurement
• z_pub_thr / z_sub_thr — unidirectional throughput measurement
• z_pub_shm_thr — SHM throughput baseline

These are synchronous programs that use .wait() rather than .await to minimize measurement overhead. That is intentional: in a benchmark you want to measure the transport, not the async runtime scheduler.

Latency: Ping/Pong

The pattern: z_ping puts a message on test/ping, z_pong subscribes and echoes it back on test/pong, and z_ping measures the round-trip time.

Run in two terminals:

# Terminal 1
cargo run --example z_pong

# Terminal 2 — 100 samples, 8-byte payload
cargo run --example z_ping -- -n 100 -s 8

Output format (per-sample RTT and derived latency):

8 bytes: seq=0 rtt=42µs lat=21µs
8 bytes: seq=1 rtt=41µs lat=20µs

The relevant code pattern from z_ping.rs — note the synchronous .wait() calls:

use zenoh::{bytes::ZBytes, key_expr::keyexpr, qos::CongestionControl, Wait};

fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(zenoh::Config::default()).wait().unwrap();

    let sub = session
        .declare_subscriber(keyexpr::new("test/pong").unwrap())
        .wait()
        .unwrap();
    let publisher = session
        .declare_publisher(keyexpr::new("test/ping").unwrap())
        .congestion_control(CongestionControl::Block)
        .express(true)
        .wait()
        .unwrap();

    let data: ZBytes = vec![0u8; 8].into();
    let write_time = std::time::Instant::now();
    publisher.put(data.clone()).wait().unwrap();
    let _ = sub.recv();
    let rtt = write_time.elapsed().as_micros();
    println!("rtt={rtt}µs lat={}µs", rtt / 2);
}

CongestionControl::Block and express(true) are set to eliminate any batching or drop-on-congestion behavior that would interfere with the latency measurement.

Throughput: pub_thr/sub_thr

z_pub_thr sends messages as fast as possible. z_sub_thr counts received messages over a 1-second window and reports throughput in messages per second and bits per second.

# Terminal 1
cargo run --example z_sub_thr

# Terminal 2 — 8-byte payload
cargo run --example z_pub_thr -- -s 8

Like z_ping, both z_pub_thr and z_sub_thr use .wait() for minimal runtime overhead. This is intentional for the same reason: the benchmark measures the transport path, not the async scheduler.

SHM Throughput

z_pub_shm_thr is the SHM variant of z_pub_thr. Use it with the standard z_sub_thr. The shared-memory feature must be enabled.

cargo run --example z_pub_shm_thr --features shared-memory -- -s 8

Typical result: SHM throughput is 3–10x higher than TCP for large payloads (above 1 KB) because no serialization or kernel copy occurs. For small payloads the benefit is smaller due to fixed per-message overhead.

Tuning Parameters

Batch Size and Low-Latency Mode

Zenoh batches small messages before sending. The default batch size is 65535 bytes. Reducing it decreases latency at the cost of throughput. The lowlatency flag disables batching entirely at the transport level:

{
  transport: {
    unicast: {
      lowlatency: true
    }
  }
}

Use lowlatency: true when round-trip latency matters more than aggregate throughput, for example in control loops or interactive systems.

Express Mode

Set express(true) on the publisher to bypass message batching for that publisher only, without a global config change:

#![allow(unused)]
fn main() {
let publisher = session
    .declare_publisher("test/ping")
    .express(true)
    .await
    .unwrap();
}

Express mode is the per-publisher equivalent of lowlatency: true. Use it for publishers that produce latency-sensitive messages alongside throughput-oriented publishers on the same session.

CongestionControl::Block

In benchmarks, use CongestionControl::Block to prevent the publisher from dropping samples when the network is saturated:

#![allow(unused)]
fn main() {
let publisher = session
    .declare_publisher("test/thr")
    .congestion_control(CongestionControl::Block)
    .await
    .unwrap();
}

In production, CongestionControl::Drop is the default and is usually correct: dropping is safer than blocking a thread. Use Block only when every sample must be delivered and back-pressure is acceptable.

Priority

For latency benchmarks, set Priority::RealTime to place messages at the front of the output queue. For throughput benchmarks the default Priority::Data is appropriate.

#![allow(unused)]
fn main() {
use zenoh::qos::Priority;

let publisher = session
    .declare_publisher("control/cmd")
    .priority(Priority::RealTime)
    .await
    .unwrap();
}

Transport Selection

Zenoh selects the best available transport automatically when both endpoints advertise the same capability. No application code change is required when the transport changes.

Debugging and Logging

This chapter covers diagnosing Zenoh applications: log levels, programmatic logging, session inspection, wire tracing, and common failure modes.

Log Levels

Zenoh uses the tracing crate. Set the log level with the ZENOH_LOG environment variable:

ZENOH_LOG=debug cargo run
ZENOH_LOG=trace cargo run

Levels in increasing verbosity: error, warn, info, debug, trace.

Start with info when debugging connectivity problems. Move to debug for routing or subscription matching issues. Use trace only when you need per-message wire-level events, as the output volume is high.

Programmatic Logging Init

Call init_log_from_env_or before opening a session. The argument is the fallback level used when ZENOH_LOG is not set:

#![allow(unused)]
fn main() {
zenoh::init_log_from_env_or("error");
}

For info as the default:

#![allow(unused)]
fn main() {
zenoh::init_log_from_env_or("info");
}

Do not call this function twice in the same process. If your application initializes its own tracing subscriber, omit this call and configure filtering for the zenoh target directly in your subscriber setup.

Session Inspection

session.info() returns the session’s own identity and its known peers and routers. Use this to verify that discovery has succeeded.

use zenoh::{session::ZenohId, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    let info = session.info();
    let zid: ZenohId = info.zid().await;
    let routers: Vec<ZenohId> = info.routers_zid().await.collect();
    let peers: Vec<ZenohId> = info.peers_zid().await.collect();

    println!("ZID:     {zid}");
    println!("Routers: {routers:?}");
    println!("Peers:   {peers:?}");
}

If routers and peers are both empty, the session has not discovered any other Zenoh node. Check multicast reachability or static peer configuration.

Wire Tracing

For SHM diagnostics, set ZENOH_TRACE_SHM=1:

ZENOH_TRACE_SHM=1 ZENOH_LOG=trace cargo run

This emits events for SHM buffer allocation, mapping, and release. Use it when SHM throughput is lower than expected or when alloc() blocks unexpectedly.

Admin Space Inspection

Query the admin space to inspect router state at runtime without modifying the application. The admin space exposes router configuration, connected peers, and active subscriptions as key-value entries.

#![allow(unused)]
fn main() {
let replies = session.get("@/router/**").await.unwrap();
while let Ok(reply) = replies.recv_async().await {
    if let Ok(sample) = reply.result() {
        let v = sample
            .payload()
            .try_to_string()
            .unwrap_or_else(|e| e.to_string().into());
        println!("{}: {}", sample.key_expr().as_str(), v);
    }
}
}

The returned values are JSON5 fragments. Parse them or pipe to jq for readable output.

Running zenohd for Debugging

Start a local router with admin space read access enabled to inspect routing without modifying your application:

zenohd --cfg 'adminspace/permissions/read:true'

Then connect your application in client mode and query the admin space. This lets you inspect active subscriptions and routing tables without instrumenting the application code.

Common Failure Modes

Interpreting Debug Output

A normal session open at info level produces output similar to:

[zenoh::net::runtime] Opening session...
[zenoh::net::runtime::orchestrator] Scouting...
[zenoh::net::runtime::orchestrator] Found peer: <ZID>
[zenoh::net::runtime] Session opened with ZID: <ZID>

If scouting finds nothing, check multicast network reachability:

ZENOH_LOG=debug cargo run 2>&1 | grep -i scout

A “Scouting” line with no “Found peer” line that follows indicates that multicast is not reaching other Zenoh nodes. Either enable multicast on the network interface, or configure a static connect endpoint in the session config:

{
  connect: {
    endpoints: ["tcp/192.168.1.10:7447"]
  }
}

Real-World Applications

This chapter covers integrating Zenoh into production systems: embedded Rust runtimes, ROS 2, V2X, edge-to-cloud telemetry, and industrial IoT.

Embedding Zenoh in an Application

The Session type is Clone and holds an internal Arc. Cloning a session is cheap: all clones share the same underlying connection state. Pass clones to independent tasks rather than wrapping the session in an additional Arc<Mutex<_>>.

use std::sync::Arc;
use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = Arc::new(zenoh::open(Config::default()).await.unwrap());

    let s1 = session.clone();
    let s2 = session.clone();

    let task1 = tokio::spawn(async move {
        s1.put("app/status", "running").await.unwrap();
    });

    let task2 = tokio::spawn(async move {
        let subscriber = s2.declare_subscriber("app/**").await.unwrap();
        while let Ok(sample) = subscriber.recv_async().await {
            println!("{}", sample.key_expr().as_str());
        }
    });

    let _ = tokio::join!(task1, task2);
}

Declare subscribers and publishers once at startup and reuse them. Declaring and dropping them in a loop is expensive: each declaration involves a network announcement.

Channel Integration

Route Zenoh samples into business logic via a Tokio channel. The callback-based subscriber allows bridging from the Zenoh callback thread to an async processing loop.

use tokio::sync::mpsc;
use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let (tx, mut rx) = mpsc::channel::<String>(32);

    let _sub = session
        .declare_subscriber("service/input")
        .callback(move |sample| {
            let msg = sample
                .payload()
                .try_to_string()
                .unwrap_or_else(|e| e.to_string().into())
                .to_string();
            let _ = tx.blocking_send(msg);
        })
        .background()
        .await
        .unwrap();

    while let Some(msg) = rx.recv().await {
        println!("Processing: {msg}");
        // business logic
    }
}

The .background() call keeps the subscriber alive for the lifetime of the session rather than tying it to a local variable. The channel backpressure (channel capacity 32) signals the callback thread to slow down if the business logic loop falls behind.

Graceful Shutdown

Use tokio::signal::ctrl_c with tokio::select to stop the application cleanly and flush the session:

use tokio::signal;
use zenoh::Config;

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();
    let subscriber = session.declare_subscriber("demo/**").await.unwrap();

    tokio::select! {
        _ = signal::ctrl_c() => {
            println!("Shutting down.");
        }
        result = async {
            while let Ok(sample) = subscriber.recv_async().await {
                println!("{}", sample.key_expr().as_str());
            }
        } => {
            let _ = result;
        }
    }

    session.close().await.unwrap();
}

Call session.close().await explicitly. This flushes pending puts and undeclares subscribers and publishers at the network level, allowing peers to update their routing tables promptly.

ROS 2: rmw_zenoh

rmw_zenoh is the Zenoh-based ROS Middleware implementation. As of ROS 2 Kilted Kaiju (2025), rmw_zenoh is a Tier-1 supported RMW. It replaces rmw_fastrtps as the default for new deployments that prioritize WAN, cross-LAN, and embedded support.

Key expression conventions in rmw_zenoh:

• Topics map to key expressions under ros2_<namespace>/<topic>
• Services map to ros2_<namespace>/<service_name>/request and .../response
• Actions use the same pattern with /goal, /result, /feedback

Set RMW_IMPLEMENTATION=rmw_zenoh_cpp before launching ROS 2 nodes. The Zenoh session config is read from $ZENOH_CONFIG or the compiled-in default config.

Benefits over DDS in ROS 2:

• No multicast flooding at scale: Zenoh uses gossip-based discovery
• WAN bridging without DDS bridges or VPNs
• SHM for intra-host camera and LiDAR traffic
• Smaller wire overhead for embedded nodes

V2X: Vehicle-to-Everything

Key expression patterns for vehicular communication:

v2x/<region>/<vehicle-id>/telemetry    — vehicle publishes position, speed
v2x/<region>/rsu/<rsu-id>/signal       — RSU publishes signal phase data
v2x/<region>/<vehicle-id>/alert        — roadside alerts to specific vehicle
v2x/<region>/**/hazard                 — broadcast hazard alerts

Vehicles declare liveliness tokens on v2x/<region>/<vehicle-id>/presence. A roadside unit monitors fleet density by subscribing to v2x/<region>/**/presence and tracking token appearances and expirations.

The liveliness pattern provides implicit fleet tracking without any application-level heartbeat protocol: Zenoh handles token expiration when a vehicle disconnects.

Edge-to-Cloud Telemetry Pipeline

Edge devices publish sensor data. A regional edge gateway stores and aggregates it. The cloud issues queries to pull current readings.

[Sensor Node] → put("plant/<id>/temp", value)
     |
     v
[Edge Gateway] — storage subscriber, retains last N samples per key
     |
     v
[Cloud] → get("plant/**/temp") with ConsolidationMode::Latest

The cloud uses ConsolidationMode::Latest to receive the most recent reading per sensor without duplicates when multiple gateways hold the same key:

#![allow(unused)]
fn main() {
use zenoh::{query::ConsolidationMode, Config};

let replies = session
    .get("plant/**/temp")
    .consolidation(ConsolidationMode::Latest)
    .await
    .unwrap();

while let Ok(reply) = replies.recv_async().await {
    if let Ok(sample) = reply.result() {
        println!("{}: {}", sample.key_expr().as_str(), sample.payload().try_to_string().unwrap_or_else(|e| e.to_string().into()));
    }
}
}

Industrial IoT: Control Loops with SHM

In a soft-PLC or motion controller pattern, three tasks communicate through Zenoh topics:

1. A sensing task reads hardware and publishes values to SHM-backed topics.
2. A control algorithm subscribes, reads the latest value via RingChannel(1), and computes actuation.
3. An actuation task subscribes to the output and drives hardware.

RingChannel(1) ensures the control algorithm always processes the most recent sample rather than accumulating a queue of stale readings.

use zenoh::{handlers::RingChannel, Config};

#[tokio::main]
async fn main() {
    zenoh::init_log_from_env_or("error");
    let session = zenoh::open(Config::default()).await.unwrap();

    // Control algorithm: always use the latest sensor value
    let sensor_sub = session
        .declare_subscriber("machine/axis/position")
        .with(RingChannel::new(1))
        .await
        .unwrap();

    let actuator_pub = session
        .declare_publisher("machine/axis/setpoint")
        .express(true)
        .await
        .unwrap();

    loop {
        if let Ok(sample) = sensor_sub.recv_async().await {
            let pos: f64 = sample.payload().deserialize().unwrap_or(0.0);
            let setpoint = compute_setpoint(pos);
            actuator_pub.put(setpoint).await.unwrap();
        }
    }
}

fn compute_setpoint(pos: f64) -> f64 {
    // PID or other control law
    -pos * 0.1
}

express(true) on the actuator publisher bypasses batching so that setpoint updates reach the actuation task with minimal added latency. Combined with SHM transport on the sensing side, the full sensing-to-actuation path involves no kernel copies and no batching delays.