Topics & Communication in Rust

This is a usage guide. For the complete API reference with per-method signatures, parameters, return types, and error conditions, see the Topic API Reference.

Topic<T> is the primary way nodes communicate in HORUS. Topics are typed, shared-memory channels that connect publishers and subscribers automatically by name.

Creating a Topic

use horus::prelude::*;

// Type annotation on the binding
let cmd: Topic<CmdVel> = Topic::new("cmd_vel")?;

// Turbofish syntax
let scan = Topic::<LaserScan>::new("scan")?;

// Custom capacity and slot size
let lidar = Topic::<LaserScan>::with_capacity("lidar/scan", 8, Some(4096))?;

Two topics with the same name and type connect automatically — one publishes, the other subscribes.

Sending Messages

use horus::prelude::*;
use horus::communication::SendBlockingError;

let topic = Topic::<CmdVel>::new("cmd_vel")?;

// Fire-and-forget. Never fails; what happens when the ring is full depends
// on the topology — see the table below.
topic.send(CmdVel::new(0.5, 0.1));

// Try without blocking — returns the message back on failure
match topic.try_send(CmdVel::new(0.5, 0.1)) {
    Ok(()) => { /* sent */ }
    Err(msg) => { /* buffer full, msg returned */ }
}

// Block up to 10ms waiting for space
match topic.send_blocking(CmdVel::new(1.0, 0.0), 10_u64.ms()) {
    Ok(()) => { /* sent */ }
    Err(SendBlockingError::Timeout) => { /* timed out */ }
}

Method	Behavior
`send(msg)`	Never returns an error. On a broadcast ring the producer overwrites the oldest slot; on a point-to-point ring it retries briefly (64 spins + 4 yields) and then drops the new message, counting it in `dropped_count()`
`try_send(msg)`	Returns `Err(msg)` if the ring is full. Point-to-point only — a broadcast ring overwrites rather than filling, so it never reports full and this never returns `Err`
`send_blocking(msg, timeout)`	Blocks until space is available or timeout elapses

Receiving Messages

use horus::prelude::*;

let topic = Topic::<Imu>::new("imu")?;

// Get next unread message (FIFO order)
if let Some(msg) = topic.recv() {
    println!("Accel: {}", msg.linear_acceleration[0]);
}

// Drain all pending messages
while let Some(msg) = topic.recv() {
    process(msg);
}

// Skip to the latest value (requires T: Copy)
if let Some(latest) = topic.read_latest() {
    println!("Latest orientation: {:?}", latest.orientation);
}

Use recv() when order matters. Messages arrive in the order they were sent and are never delivered twice. Use read_latest() for state-like data (sensor readings, poses) where only the newest value matters.

⚠️recv() preserves order, not delivery

recv() does not guarantee you see every message. On a broadcast POD topic (PodShm — a fixed-size message with more than one subscriber, which is the usual shape for a camera or IMU feed), a producer that laps a slow consumer overwrites slots the consumer has not read yet. The consumer detects this and jumps forward — for a drain loop, usually straight to the current head — and everything it skipped is gone. The recv() that noticed the lap returns None, so a while let Some(..) drain ends early on that cycle rather than reporting anything.

The skip is silent. It raises no error and increments no counter — dropped_count() reports send-side failures only, so it stays at 0 while a slow subscriber is losing frames.

send_blocking() does not help here. It applies backpressure by retrying a try_send that failed, and a broadcast ring never fails a send — it overwrites — so the first attempt succeeds and the call returns in nanoseconds while the slow subscriber is lapped exactly as before. Backpressure only exists on a point-to-point ring.

What does work on a broadcast topic: keep the consumer fast enough to drain the ring, give the topic more capacity so a transient stall does not lap it, or move the slow work off the subscriber's tick.

Zero-Copy for Large Data

For Image, PointCloud, and DepthImage, HORUS uses pool-backed allocation. Only a descriptor goes through the ring buffer — the payload stays in shared memory.

use horus::prelude::*;

let camera = Topic::<Image>::new("camera/rgb")?;

// Send — moves the Image into the pool slot
camera.send(image);

// Receive — zero-copy access to the image data
if let Some(img) = camera.recv() {
    println!("{}x{} image received", img.width(), img.height());
}

The same API works for other pool-backed types:

let cloud_topic = Topic::<PointCloud>::new("lidar/points")?;
let depth_topic = Topic::<DepthImage>::new("camera/depth")?;

Custom Message Types

Define your own messages with the message! macro:

use horus::prelude::*;

message! {
    #[fixed]
    /// Motor feedback at 1kHz
    MotorFeedback {
        motor_id: u32,
        velocity: f32,
        current_amps: f32,
        temperature_c: f32,
    }
}

let feedback = MotorFeedback::topic("motor.feedback")?;
feedback.send(MotorFeedback {
    motor_id: 1,
    velocity: 3.14,
    current_amps: 0.5,
    temperature_c: 45.0,
});

The macro generates serialization traits automatically. The #[fixed] attribute — written before the doc comment, directly above the message name — emits the struct as #[repr(C)] and Copy, which is what puts it on the zero-copy fast path: the bytes are written straight into the ring slot with no serialization step. Leave #[fixed] off for messages whose fields are not all Copy (anything containing a String or Vec); those go through a bincode serialize/deserialize round trip on every send and receive. Which shared-memory backend carries a topic is then chosen automatically from its producer and consumer counts — see Benchmarks for measured per-route latencies.

Guard against a message that changed shape

Topic::new checks the message type's name and, for fixed messages, its size. Neither says anything about field layout, so two builds of the same message that keep the name and the size while reordering fields will happily share a topic:

// Robot A, built from v1.0 of the message crate
message! { #[fixed] Pose { x: f32, y: f32 } }

// Robot B, built from v1.1 — someone reordered the fields
message! { #[fixed] Pose { y: f32, x: f32 } }

Same name, same eight bytes. A sends (x=1, y=2) and B receives Pose { y: 1.0, x: 2.0 } — the coordinates arrive swapped, with no error anywhere. Halfway through a fleet rollout, that is a robot driving the wrong way.

Every message! type carries a LAYOUT_HASH covering its name and each field's name and type, and a Type::topic(name) helper that supplies it:

let pose = Pose::topic("robot.pose")?;        // layout-checked
let pose = Topic::<Pose>::new("robot.pose")?; // unchecked

With the checked form, B's open fails instead:

Communication error: Failed to create topic 'robot.pose': message layout
mismatch. This build's 'msgs::Pose' hashes to 0xf5f319cf, but the topic was
opened with 0x1ab36d8b.
The type name and size match, so only the field layout differs — two builds of
the same message that reordered, renamed or retyped a field. Reading it would
silently reinterpret the bytes rather than fail.
Fix: rebuild both sides against the same message definition.

Prefer Type::topic(...). It costs nothing at runtime — the hash is a compile- time constant and is compared once, when the topic is opened. A hash of zero means "not supplied", so a peer still using Topic::new is never rejected; it is simply not protected.

Using Topics in Nodes

Topics are typically stored as fields on your node struct:

use horus::prelude::*;

struct ObstacleAvoider {
    scan_in: Topic<LaserScan>,
    cmd_out: Topic<CmdVel>,
}

impl ObstacleAvoider {
    fn new() -> Result<Self> {
        Ok(Self {
            scan_in: Topic::new("scan")?,
            cmd_out: Topic::new("cmd_vel")?,
        })
    }
}

impl Node for ObstacleAvoider {
    fn name(&self) -> &str { "obstacle_avoider" }

    fn tick(&mut self) {
        if let Some(scan) = self.scan_in.recv() {
            let min_range = scan.ranges.iter()
                .copied()
                .filter(|r| *r > scan.range_min && *r < scan.range_max)
                .fold(f32::MAX, f32::min);

            let speed = if min_range < 0.5 { 0.0 } else { 1.0 };
            self.cmd_out.send(CmdVel::new(speed, 0.0));
        }
    }
}

Communication Patterns

One-to-many — one publisher, multiple subscribers. Each subscriber reads the stream independently, so a slow one cannot starve the others (though it can still be lapped — see Receiving Messages):

// Publisher node
let status = Topic::<Heartbeat>::new("status")?;
status.send(Heartbeat::new("status_publisher", 1));

// Subscriber A and B both receive the heartbeat
let status_a = Topic::<Heartbeat>::new("status")?;
let status_b = Topic::<Heartbeat>::new("status")?;

Many-to-one — multiple publishers write to the same topic. The subscriber sees all messages interleaved:

// Motor 1 and Motor 2 both publish to "motor.feedback"
let fb1 = Topic::<MotorFeedback>::new("motor.feedback")?;
let fb2 = Topic::<MotorFeedback>::new("motor.feedback")?;

fb1.send(MotorFeedback { motor_id: 1, velocity: 3.0, current_amps: 0.4, temperature_c: 42.0 });
fb2.send(MotorFeedback { motor_id: 2, velocity: 3.1, current_amps: 0.5, temperature_c: 44.0 });

// Aggregator reads all feedback
let all_fb = Topic::<MotorFeedback>::new("motor.feedback")?;
while let Some(fb) = all_fb.recv() {
    println!("Motor {} velocity: {}", fb.motor_id, fb.velocity);
}

Monitoring

Check topic health at runtime:

let topic = Topic::<CmdVel>::new("cmd_vel")?;

if topic.has_message() {
    println!("{} messages pending", topic.pending_count());
}

if topic.dropped_count() > 0 {
    hlog!(warn, "Dropped {} messages", topic.dropped_count());
}

let m = topic.metrics();
println!("Sent: {}, Received: {}", m.messages_sent(), m.messages_received());

⚠️`messages_sent` / `messages_received` read 0 in normal operation

These two counters are incremented only on the verbose content-logging path, which is #[cold] and runs only while a topic's verbose flag is set from the horus monitor TUI. The ordinary send() and recv() hot paths do not touch them, so in a normal run both report 0 no matter how much traffic the topic carries.

What is always maintained is the per-topic messages_total counter in the shared-memory header, incremented on every send() regardless of backend. That is what horus topic list and horus topic hz report.

It counts send() only — try_send() and send_blocking() do not touch it. A topic driven exclusively by those reads as 0 messages and 0 Hz in both commands while carrying full traffic, so do not use them to decide a critical-command topic is dead. dropped_count() and pending_count() are also always live — with the caveat that dropped_count() counts send-side failures only (see Receiving Messages).