Topic and Pub/Sub

Key Takeaways

After reading this guide, you will understand:

• How Topic provides ultra-fast pub/sub communication (~75ns in-process, ~200-300ns cross-process) with automatic optimization
• The send() and recv() methods for publishing and subscribing to typed topics
• Communication patterns (one-to-one, one-to-many, many-to-one, many-to-many) for different architectures
• When to use Topic for real-time, single-machine communication vs network-based messaging

The Topic is HORUS's ultra-low latency publish-subscribe (pub/sub) communication system. It enables nodes to exchange messages through shared memory IPC with automatic optimization — ~75ns median for a 16-byte message between threads in one process, ~200-300ns between separate processes, depending on topology.

What is a Topic?

A Topic<T> is a typed communication channel that connects publishers and subscribers. The system automatically detects the optimal backend based on:

• How many publishers and subscribers exist
• Whether the message type is simple and fixed-size or needs serialization

You just call send() and recv() — HORUS handles the rest. Participants may share a thread, share a process, or live in different processes; every topic is shared-memory backed either way, so location changes neither the API nor the backend choice.

Key Features

Automatic Optimization: The fastest communication path is auto-detected at runtime

Zero-Copy Communication: Simple fixed-size types are shared directly in memory without serialization

Type Safety: Compile-time guarantees for message types — within one program. Whether a publisher and subscriber in different nodes, processes or languages agree is checked at runtime, when the topic is opened

Live Migration: paths upgrade immediately when a participant joins. A departure is only noticed when that participant's 5-second lease is later reclaimed by a new registration, so a topic can stay on the wider backend for a while after a subscriber exits

Automatic Optimization

HORUS auto-selects the fastest communication path based on your topology. You never need to configure this — it happens automatically when participants call send() or recv().

The message type matters too: simple fixed-size types get a zero-copy ring, everything else gets a serializing one. The rings are internal — there is no backend type to name, import or select. See Performance for measured latency.

If topology changes (e.g., a second subscriber joins), HORUS automatically migrates to the optimal path without dropping messages.

// All of these use the same API — optimization is automatic
let topic: Topic<f32> = Topic::new("velocity")?;
topic.send(1.5);
let msg = topic.recv();

Tip: Simple fixed-size structs (no String, Vec, Box) automatically get a faster zero-copy path. Both types work with the same API — HORUS picks the fastest path for you.

Basic Usage

Creating a Topic

use horus::prelude::*;

// Create a Topic for f32 values on topic "velocity"
let topic: Topic<f32> = Topic::new("velocity")?;

Publishing Messages

use horus::prelude::*;

struct Publisher {
    velocity_pub: Topic<f32>,
}

impl Publisher {
    fn new() -> Result<Self> {
        Ok(Self {
            velocity_pub: Topic::new("velocity")?,
        })
    }
}

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

    fn tick(&mut self) {
        let velocity = 1.5;

        // Send message — infallible, non-blocking
        self.velocity_pub.send(velocity);
    }
}

Subscribing to Messages

use horus::prelude::*;

struct Subscriber {
    velocity_sub: Topic<f32>,
}

impl Subscriber {
    fn new() -> Result<Self> {
        Ok(Self {
            velocity_sub: Topic::new("velocity")?,
        })
    }
}

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

    fn tick(&mut self) {
        if let Some(velocity) = self.velocity_sub.recv() {
            println!("Received velocity: {}", velocity);
        }
    }
}

Transport

Topics use local shared memory for ultra-fast communication between processes on the same machine.

// Automatically uses local shared memory
let topic: Topic<SensorData> = Topic::new("sensors")?;

• Performance: ~75ns median for a 16-byte message in one process, ~200-300ns across processes
• Use case: All nodes on same machine
• Characteristics: Ultra-fast, deterministic, zero-copy

For multi-machine communication, HORUS supports zero-configuration peer discovery over UDP multicast (group 224.0.69.72, port 9100). Enable it with the net feature flag: horus = { version = "0.2", features = ["net"] }. Override discovery with HORUS_NET_MULTICAST, HORUS_NET_PORT, or HORUS_NET_PEER=ip1,ip2 for direct unicast.

API Reference

Note: HORUS separates introspection from the hot path. The send() and recv() methods have zero logging overhead by default.

send()

pub fn send(&self, msg: T)

Publishes a message to the topic. Fire-and-forget — it returns nothing, so there is no Result to handle, but delivery is not guaranteed. On point-to-point topologies (1 publisher / 1 subscriber, or many publishers / 1 subscriber) a full ring rejects the new message: send() retries briefly (spin, then yield) and then drops it; queued messages are preserved in FIFO order. On broadcast topologies (any number of publishers, multiple subscribers) the ring is latest-wins and the oldest slot is overwritten instead. Use try_send() to detect the drop, or send_blocking(msg, timeout) to wait for room on critical topics.

// send() returns no Result — but the message can still be dropped
topic.send(data);

recv()

pub fn recv(&self) -> Option<T>

Receives the next message from the topic. Returns None if no message is available. Non-blocking.

if let Some(data) = topic.recv() {
    // Process the received message
}

try_send()

pub fn try_send(&self, msg: T) -> Result<(), T>

Attempts to send a message, returning it on failure. Unlike send(), which on point-to-point topologies retries briefly and then drops the message, try_send() hands it back to you. On broadcast topologies the ring is latest-wins and never reports full, so try_send() does not fail there.

match topic.try_send(data) {
    Ok(()) => { /* sent successfully */ }
    Err(returned_data) => { /* point-to-point ring full, message returned */ }
}

send_blocking()

pub fn send_blocking(&self, msg: T, timeout: std::time::Duration) -> Result<(), SendBlockingError>

Sends a message, waiting for room in the ring instead of dropping it. It tries once immediately, then spins, then yields, then sleeps in 100µs steps until the deadline. The only error is SendBlockingError::Timeout, returned when the ring stayed full for the whole timeout — the message is not handed back. Use this on critical topics (emergency stop, motor setpoints) where a dropped message is unacceptable.

It only does anything on a point-to-point ring. A broadcast ring (multiple subscribers) never reports a full ring — it overwrites the oldest slot — so the first internal attempt always succeeds and send_blocking returns immediately, applying no backpressure at all.

SendBlockingError is not in the prelude; import it from horus::communication::SendBlockingError if you need to match on it.

use std::time::Duration;

if topic.send_blocking(command, Duration::from_millis(5)).is_err() {
    // ring stayed full for 5ms — escalate rather than continue silently
}

read_latest()

pub fn read_latest(&self) -> Option<T>
where
    T: Copy,

Returns the most recent message without advancing the consumer position. Calling it multiple times returns the same message until a new one is published. Useful for reading infrequently-updated data like static transforms or configuration.

T: Copy requirement: read_latest() requires T: Copy to guarantee safe concurrent reads. Types with heap allocations (String, Vec, etc.) should use recv() instead.

// Read latest transform (doesn't consume it)
if let Some(transform) = tf_topic.read_latest() {
    self.apply_transform(transform);
}

has_message()

pub fn has_message(&self) -> bool

Checks if at least one message is available without consuming it.

pending_count()

pub fn pending_count(&self) -> u64

Returns the number of messages waiting to be consumed.

Runtime Debug Logging

Debug logging is toggled at runtime from the TUI monitor — no code changes or recompilation needed. Select a topic in the Topics tab and press Enter to start logging; press Esc to stop.

When debug logging is active, every send() and recv() records a metadata entry — topic name, direction, and IPC latency — to the log buffer. The Rust transport never calls log_summary(), so implementing LogSummary does not change what a topic logs; use it yourself inside hlog! when you want the message contents in the log.

horus monitor --tui    # Open TUI, navigate to Topics tab, press Enter on a topic

Debug logging adds no overhead when disabled — introspection is fully separated from the hot path.

with_capacity()

pub fn with_capacity(name: &str, capacity: u32, slot_size: Option<usize>) -> Result<Self>

Creates a topic with a custom ring buffer capacity. By default, HORUS auto-sizes capacity based on the message type size. Use this when you need more buffering (e.g., bursty producers).

// 256-slot ring buffer (rounded up to next power of 2)
let topic: Topic<f32> = Topic::with_capacity("velocity", 256, None)?;

Type-Safe Topic Descriptors

The topics! macro defines compile-time topic descriptors that prevent topic name typos and keep topic names consistent across your codebase.

Defining Topics

use horus::prelude::*;

// `topics!` is not in the prelude — invoke it through the re-exported crate
horus::topics! {
    pub CMD_VEL: CmdVel = "cmd_vel",
    pub SENSOR_DATA: SensorReading = "sensor.data",
    pub MOTOR_STATUS: MotorState = "motor.status",
}

Using Descriptors

// Publisher — the descriptor supplies the name; the message type is chosen here
let pub_topic: Topic<CmdVel> = Topic::new(CMD_VEL.name())?;
pub_topic.send(CmdVel::new(1.0, 0.5));

// Subscriber — the descriptor supplies the name; the message type is chosen here
let sub_topic: Topic<CmdVel> = Topic::new(CMD_VEL.name())?;
if let Some(cmd) = sub_topic.recv() {
    // cmd is guaranteed to be CmdVel
}

This prevents common errors like:

• Typos in topic names (caught at compile time)
• Inconsistent topic names between publisher and subscriber, or across modules
• Topic names drifting out of sync when one call site is renamed

The declared message type in topics! is documentation only — TopicDescriptor<T> carries T as PhantomData and name() returns a plain &'static str, so each call site still picks its own Topic<T>.

Communication Patterns

The API is identical in every pattern — only the ring HORUS picks underneath changes. Measured latency for each topology lives in Performance below.

One-to-One

Single publisher, single subscriber — a point-to-point ring, FIFO.

struct PubNode {
    data_pub: Topic<f32>,
}

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

    fn tick(&mut self) {
        self.data_pub.send(42.0);
    }
}

struct SubNode {
    data_sub: Topic<f32>,
}

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

    fn tick(&mut self) {
        if let Some(data) = self.data_sub.recv() {
            println!("Got: {}", data);
        }
    }
}

One-to-Many (Broadcast)

Single publisher, multiple subscribers — a broadcast ring, so each subscriber reads the stream independently rather than competing for messages. One slow subscriber does not starve the others.

That is independence, not a delivery guarantee. The ring is latest-wins: a publisher that laps a subscriber overwrites slots that subscriber has not read, and the subscriber jumps forward, usually to the current head, and the skipped messages are lost. The loss is silent — dropped_count() counts send-side failures, and on a broadcast ring the publisher never fails to send, so it stays at 0. Keep subscribers fast enough to drain the ring, or use send_blocking() where a gap is unacceptable.

// One publisher
struct Broadcaster {
    broadcast_pub: Topic<String>,
}

// Multiple subscribers — each reads the stream independently
struct Listener1 { broadcast_sub: Topic<String> }
struct Listener2 { broadcast_sub: Topic<String> }
struct Listener3 { broadcast_sub: Topic<String> }

Many-to-One (Aggregation)

Multiple publishers, single subscriber — a multi-producer point-to-point ring, FIFO across all publishers.

// Multiple publishers
struct Sensor1 { reading_pub: Topic<f32> }
struct Sensor2 { reading_pub: Topic<f32> }

// Single aggregator
struct Aggregator {
    reading_sub: Topic<f32>,
}

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

    fn tick(&mut self) {
        if let Some(reading) = self.reading_sub.recv() {
            self.process(reading);
        }
    }
}

Many-to-Many

Multiple publishers and subscribers — a broadcast ring, same as one-to-many.

struct Agent1 {
    state_pub: Topic<RobotState>,
    state_sub: Topic<RobotState>,
}

struct Agent2 {
    state_pub: Topic<RobotState>,
    state_sub: Topic<RobotState>,
}

Topic Naming

Use Dots, Not Slashes

Important: HORUS uses dots (.) for topic hierarchy, not slashes (/).

// CORRECT - Use dots
let topic = Topic::new("sensors.lidar");
let topic = Topic::new("robot.cmd_vel");

// DISCOURAGED - slashes work, but break the convention
let topic = Topic::new("sensors/lidar");   // Discouraged — works, but breaks the dot convention
let topic = Topic::new("robot/cmd_vel");   // Discouraged — works, but breaks the dot convention

Why dots instead of slashes?

Dots are the HORUS convention; slashes are accepted but produce nested directories under /dev/shm/horus_<ns>/topics/. A leading slash (ROS-style /sensor/lidar) is rejected as an absolute path.

Coming from ROS?

Best Practices

// Descriptive names
let topic = Topic::new("cmd_vel");           // Good
let topic = Topic::new("data");              // Too vague

// Hierarchical naming
let topic = Topic::new("sensor.lidar");      // Hierarchical
let topic = Topic::new("robot1.cmd_vel");    // Namespaced
let topic = Topic::new("diagnostics.cpu");   // Categorized

Reserved Topic Names

Avoid using these patterns:

• Topics starting with _ (internal use)
• Topic names beginning with / (rejected as an absolute path)
• Topics with special characters: !@#$%^&*()

Error Handling

Send Behavior

send() is fire-and-forget — it returns nothing, so there is no Result to handle, but delivery is not guaranteed. On point-to-point topologies (1 publisher / 1 subscriber, or many publishers / 1 subscriber) a full ring rejects the new message: send() retries briefly (spin, then yield) and then drops it; queued messages are preserved in FIFO order. On broadcast topologies (any number of publishers, multiple subscribers) the ring is latest-wins and the oldest slot is overwritten instead.

// send() returns no Result — but the message can still be dropped
topic.send(data);

For explicit error handling, use try_send():

match topic.try_send(data) {
    Ok(()) => { /* success */ }
    Err(msg) => { /* buffer full — msg returned to caller */ }
}

Receive Behavior

recv() returns None when no message is available — this is normal, not an error:

match topic.recv() {
    Some(data) => {
        // Process data
    }
    None => {
        // No data available - this is normal
    }
}

Message Types

What Types Can I Use?

Most types work with Topic — just add standard derives:

use serde::{Serialize, Deserialize};

// Primitive types work out of the box
let topic: Topic<f32> = Topic::new("float_topic")?;
let topic: Topic<bool> = Topic::new("bool_topic")?;

// Custom structs — just add derives
#[derive(Clone, Serialize, Deserialize)]
struct MyMessage {
    x: f32,
    y: f32,
    name: String,
}

let topic: Topic<MyMessage> = Topic::new("my_topic")?;

// Collections work too
#[derive(Clone, Serialize, Deserialize)]
struct SensorBatch {
    readings: Vec<f32>,
    timestamp: u64,
}

let topic: Topic<SensorBatch> = Topic::new("batch_topic")?;

Required Traits

Your message types need these traits (all auto-derived):

Additionally, types must be Send + Sync + 'static — satisfied automatically by most types.

use serde::{Serialize, Deserialize};

#[derive(Clone, Serialize, Deserialize)]
struct MyMessage {
    // your fields
}

Advanced Usage

Conditional Publishing

Only publish when certain conditions are met:

impl Node for ConditionalPublisher {
    fn tick(&mut self) {
        let data = self.read_sensor();

        if data > self.threshold {
            self.alert_pub.send(data);
        }
    }
}

Message Buffering

Cache the last received message using read_latest() or manual buffering:

struct BufferedSubscriber {
    data_sub: Topic<f32>,
    last_value: Option<f32>,
}

impl Node for BufferedSubscriber {
    fn tick(&mut self) {
        if let Some(value) = self.data_sub.recv() {
            self.last_value = Some(value);
        }

        if let Some(value) = self.last_value {
            self.process(value);
        }
    }
}

For Copy types, use read_latest() instead — it reads the latest message without consuming it:

// Requires T: Copy — reads latest without advancing consumer position
if let Some(transform) = self.tf_sub.read_latest() {
    self.apply_transform(transform);
}

Rate Limiting

Publish at a specific rate:

struct RateLimitedPublisher {
    data_pub: Topic<f32>,
    tick_count: u32,
    publish_every_n_ticks: u32,
}

impl Node for RateLimitedPublisher {
    fn tick(&mut self) {
        self.tick_count += 1;

        if self.tick_count % self.publish_every_n_ticks == 0 {
            self.data_pub.send(42.0);
        }
    }
}

Message Filtering

Filter messages before processing:

impl Node for FilteringSubscriber {
    fn tick(&mut self) {
        if let Some(data) = self.data_sub.recv() {
            if data.is_valid() && data.quality > 0.8 {
                self.process(data);
            }
        }
    }
}

Memory and Capacity

Memory Usage Per Topic

By default, Topic::new() auto-sizes capacity based on message type size:

Auto capacity is always clamped to [16, 1024] and rounded up to the next power of two. Slot size tracks the message size for simple fixed-size types; only non-POD (heap-owning) types fall back to the 8 KB default slot.

For most robotics applications, memory usage per topic is under 1 MB.

Cleaning Up

Shared memory files persist after processes exit (by design — allows new processes to join existing topics). Clean up between sessions:

# Clean shared memory only (recommended)
horus clean --shm

# Preview what would be cleaned
horus clean --shm --dry-run

# Clean everything (shared memory + build cache)
horus clean --all

A stale topic file left by a dead process is reused, not deleted: its header is re-initialised when a live process next opens the same topic. Nothing removes the file itself, so /dev/shm accumulates entries across runs — clear them with horus clean --shm.

Troubleshooting

"No space left on device" — Shared memory is full:

horus clean --shm            # Clean up

Type mismatch — Ensure publisher and subscriber use the exact same type for a topic name:

// Both sides MUST use the same type
let pub_topic: Topic<CmdVel> = Topic::new("cmd_vel")?;
let sub_topic: Topic<CmdVel> = Topic::new("cmd_vel")?;

Performance

Latency by Topology (16B message)

Every topic is shared-memory backed, so these are end-to-end one-way IPC figures, not in-process function calls. Measured by all_paths_latency on an Intel Core i7-10750H (6C/12T, powersave governor), 100K samples per scenario:

The ~167ns row is the bare atomic round of a shared-memory handoff with no framework on top — a bound HORUS approaches, not a latency any topic achieves.

Latency by Message Type

Per-message-type figures come from robotics_messages_benchmark (same process, publisher and subscriber on separate threads, 50K iterations). CmdVel (16B) is the reference point at ~75ns median / ~135ns p99; see Benchmarks for the other message types.

Latency grows with message size, with one exception: Image, PointCloud, DepthImage, OccupancyGrid and CostMap are pool-backed — only a small fixed-size descriptor (224-440 bytes, depending on the type) crosses the ring, so their latency is independent of payload size.

Throughput

• Millions of messages per second for small messages
• Gigabytes per second for large messages
• Deterministic latency regardless of system load

Best Practices

Use simple fixed-size types when possible — they get the fastest path automatically:

Topic::<[f32; 3]>::new("position")?;   // Fixed-size: fastest path
Topic::<Vec<f32>>::new("position")?;   // Dynamic: still fast, but uses serialization

Keep messages small — latency grows with size.

Check recv() every tick — don't skip ticks:

fn tick(&mut self) {
    if let Some(msg) = self.sub.recv() {
        self.process(msg);
    }
}

Use topics! macro for shared topic definitions across modules:

horus::topics! {
    pub CMD_VEL: CmdVel = "cmd_vel",
    pub ODOM: Odometry = "odom",
}

send() returns no Result — but delivery is not guaranteed, so reach for try_send() or send_blocking() on critical topics:

self.telemetry_pub.send(sample);              // best-effort — may be dropped under load

if self.cmd_pub.try_send(command).is_err() {
    // ring full — the command was returned, handle or retry it
}

Next Steps

• Learn about the Scheduler for orchestrating nodes
• Explore Message Types for standard robotics messages
• Read the API Reference for complete Topic documentation