Pub/Sub

An async Publish/Subscribe API running directly over Zenoh peers by default.

While Bagfile Query and Variable Sharing are the main features of Minot, this library is just a small cherry on top. The mt_pubsub library uses existing networking primitives from Minot (like dynamic Rules and Ships) and implements a minimal Publish/Subscribe pattern on top of it.

When used in static and simple environments, it is stable enough for replacing the ROS2 Publish/Subscribe API. But it is absolutely not on par with ROS features.

Several Stacks on One Machine

A program that starts a whole stack of its own (Pelorus replaying a bag, with the Minot it spawns) can take a network no other stack on the machine uses:

let claim = mt_pubsub::claim_network()?;
// Before any thread starts, so children and this process's sessions agree:
unsafe { claim.apply_to_process() };

The first claim gets the defaults (7447, domain 0), so tools left on them still see it. Later ones get a free port, with the same number as their domain. Claims are locked per user in the temp directory and released when dropped or when the process ends. A process that finds MINOT_ZENOH_PORT already set keeps it and claims nothing, which is how a parent hands its choice to its children. To reach a stack that is not on the defaults from a viewer, start the viewer with the same MINOT_ZENOH_PORT and MINOT_DOMAIN_ID.

Peer Discovery

NodeConfig::new selects CoordMode::Peer. Nodes declare typed topics directly in Zenoh without starting a Minot coordinator or acquiring a lock file. Multicast scouting and single-hop gossip connect peers automatically. MINOT_ZENOH_PORT (default 7447) selects the peer UDP scouting port and the existing coordinator TCP port; use a different value on all Minot nodes to coexist with ROS Zenoh. Peer listeners use OS-assigned TCP ports unless explicitly configured.

For Docker or routed networks without multicast, supply reachable peers through MINOT_PEERS=tcp/seed:17447 (comma-separated endpoints), or use NodeConfig::peer_endpoints and NodeConfig::listen_endpoint. Explicit peers disable multicast; gossip remains enabled. Bootstrap nodes are ordinary peers. They must advertise reachable listener addresses for mesh expansion.

Discovery is asynchronous. Use publisher.wait_for_subscribers(Duration::from_secs(5)).await? when at least one recipient must be discovered before publishing the first sample. For a known startup group, wait_for_subscriber_count(count, timeout) waits for the required number of receiving endpoints. Startup QoS checks and counts use the peers observed so far if a discovery query times out during mesh expansion; Reliable delivery still requires a complete subscriber snapshot and all ACKs. Local-only peers choose one connection initiator per pair to avoid duplicate connection attempts.

Coordinated Compatibility

mt_rat variable synchronization and bag-query coordination retain their coordinator protocol. Pub/sub applications can explicitly select CoordMode::AutoStart, Start, or External to use the previous topic routing. Peer and coordinated pub/sub endpoints use different wire protocols; select the same mode on both sides of a topic.

The coordinator's listener and local startup probe honor MINOT_ZENOH_PORT. Explicit MINOT_COORD_ADDR keeps its own port. Its per-user/per-port lock still requires a shared filesystem; use one designated coordinator for coordinated applications across separate containers.

Peer Reliable delivery acknowledges subscriber queue acceptance. Missing acknowledgements or loss of a fatal reliable node cancels the peer run. Transient-local writers provide retained samples directly to late subscribers. Peer loss detection uses Zenoh liveliness; coordinator heartbeat/reconnect settings continue to apply to coordinated modes. The peer implementation uses owned Zenoh buffers; explicit Minot SHM pool controls remain on the coordinated path.

Examples

After adding the library to your Cargo.toml, you can create a publisher and a subscriber.

use mt_pubsub::{Node, NodeConfig, Qos};
use ros2_interfaces_jazzy_rkyv::std_msgs::msg;
use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Opens a coordinator-free Zenoh peer session
    let node = Node::create(NodeConfig::new("my_node_pub")).await?;     

    // Qos can be Reliable or BestEffort
    let pubber = node
        .create_publisher::<msg::String>("/some_topic".to_owned(), Qos::Reliable)
        .await?;

    let msg = msg::String {
        data: "Hello World 🖖".to_owned(),
    };

    loop {
        pubber.publish(&msg).await?;
        sleep(Duration::from_secs(1)).await;
    }
}

In contrast to Variable Sharing, the publisher does not care about syncing with the subscriber. But like most abstractions, this comes at a cost. Here it is setting the capacity of the queue length for the subscriber - like in ROS.

use mt_pubsub::{Node, NodeConfig, Qos};
use ros2_interfaces_jazzy_rkyv::std_msgs::msg;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let node = Node::create(NodeConfig::new("my_node_sub")).await?;

    let mut subber = node
        .create_subscriber::<msg::String>("/some_topic".to_owned(), 10, Qos::Reliable)
        .await?;

    while let Some(msg) = subber.next().await {
        println!("Received Message: {}", &msg.data);
    }


    Ok(())
}

Mind that we don't need to spin_once here thanks to async like in other Rust ROS libraries.