Topic API

Topics are how nodes communicate in HORUS. Topic<T> provides typed publish/subscribe channels backed by shared memory for zero-copy IPC. Two Topic instances with the same name and type connect automatically — one publishes, the other subscribes. Every topic is shared-memory-backed from creation, so a cross-process consumer can always join later with no migration. HORUS picks the matching SHM backend automatically from the pub/sub pattern — SpscShm (1:1), MpscShm (N:1), PodShm (1:N POD broadcast), or FanoutShm (N:M / non-POD broadcast) — there is no separate in-process path.

Python: Available via horus.Topic("name", Type). See Python Bindings.

New to topics? Start with Topics: How Nodes Talk for a 5-minute introduction.

// simplified
use horus::prelude::*;

Quick Reference — Constructors

Quick Reference — Sending

Quick Reference — Receiving

Quick Reference — State & Metrics

Constructor Methods

new(name)

Creates a topic with default capacity and auto-sized slots.

Signature

// simplified
pub fn new(name: impl Into<String>) -> HorusResult<Self>

Parameters

Returns

HorusResult<Topic<T>> — connected to the named shared memory channel.

Errors

Behavior

• Default capacity: auto-computed (16–1024 slots based on message size; smaller messages get more slots)
• Default slot size: size_of::<T>() rounded up to page alignment
• Backend auto-selection: HORUS picks the shared-memory backend from the topic's pub/sub pattern — SpscShm (1:1), MpscShm (N:1), PodShm (1:N POD broadcast), or FanoutShm (N:M / non-POD broadcast). Every backend is SHM-backed from creation, so a cross-process subscriber can join later with no migration.
• Creating two Topic<T> with the same name connects them — no broker, no configuration

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

let pub_topic = Topic::<CmdVel>::new("cmd_vel")?;
let sub_topic = Topic::<CmdVel>::new("cmd_vel")?; // connects to same channel

pub_topic.send(CmdVel::new(1.0, 0.0));
let msg = sub_topic.recv(); // Some(CmdVel { linear: 1.0, angular: 0.0 })

with_capacity(name, capacity, slot_size)

Creates a topic with explicit buffer configuration.

Signature

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

Parameters

Returns

HorusResult<Topic<T>>

When to use

• Large messages that need bigger slots (e.g., LaserScan with 360+ ranges)
• High-frequency topics where you need more buffering to prevent drops
• Small messages where you want to reduce memory footprint

When NOT to use

• Most cases — Topic::new() auto-sizes correctly
• Pool-backed types (Image, PointCloud, DepthImage, Tensor) — they manage their own memory

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

// 8 slots, 4KB each — for large LiDAR scans
let scan = Topic::<LaserScan>::with_capacity("lidar.scan", 8, Some(4096))?;

// 16 slots, auto-sized — high-frequency IMU with extra buffering
let imu = Topic::<Imu>::with_capacity("imu", 16, None)?;

Sending Methods

send(msg)

Publishes a message. Non-blocking. If the buffer is still full after a brief spin+yield retry, the newly-sent message is dropped (a slow consumer keeps the older, stale data).

Signature

// simplified
pub fn send(&self, msg: T)

Parameters

Returns

Nothing (()). Never returns an error, but the message may be dropped if the buffer stays full (see Behavior). For guaranteed delivery use send_blocking().

Behavior

• If all slots stay occupied through a brief spin+yield retry, the newly-sent message is silently dropped (the older, unread data is kept)
• Use dropped_count() to detect drops
• Notifies any event-driven nodes (.on("topic")) that new data arrived
• Latency: every topic is SHM-backed from creation (so a cross-process consumer can join later), so send() always writes to the shared-memory ring. The cost floor is core locality, not intra-vs-SHM: with publisher and subscriber on the same executor thread (same core, L1-hot) a send+recv round trip is ~39 ns (~19 ns one-way); when the consumer runs on a different core — whether another thread or another process — it is ~220 ns one-way, which is cache-coherence physics that no backend or layout choice can avoid.

When to use

• Default for real-time robotics — you always want the latest data
• Control loops where blocking is unacceptable
• Fire-and-forget telemetry

When NOT to use

• When you need to know if the message was received — use try_send() instead
• When dropping messages is unacceptable — use send_blocking() instead

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

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

// Fire-and-forget (non-blocking; drops the new message if the buffer is full)
topic.send(CmdVel::new(1.0, 0.0));

// Check if messages were dropped
if topic.dropped_count() > 0 {
    hlog!(warn, "{} messages dropped — subscriber too slow", topic.dropped_count());
}

try_send(msg)

Attempts to publish without overwriting. Returns the message back if the buffer is full.

Signature

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

Parameters

Returns

• Ok(()) — message published successfully
• Err(msg) — buffer full, message returned to caller (not consumed)

Behavior

• Non-blocking. Checks buffer space and either publishes or returns immediately.
• The returned message can be retried or discarded — your choice.

When to use

• When you need to detect drops — implement backpressure, count losses, or log warnings
• When the message is expensive to create and you don't want to waste it

When NOT to use

• Normal control loops — use send() (simpler, always succeeds)

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

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

match topic.try_send(CmdVel::new(1.0, 0.0)) {
    Ok(()) => { /* sent */ }
    Err(_returned) => {
        hlog!(warn, "Buffer full — message not sent");
    }
}

send_blocking(msg, timeout)

Blocks until buffer space is available or the timeout elapses.

Signature

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

Parameters

Returns

• Ok(()) — message published
• Err(SendBlockingError::Timeout) — buffer stayed full for the entire timeout

Behavior

Uses a graduated wait strategy for low latency:

1. Immediate try_send (0 latency)
2. Spin loop — 256 iterations (~sub-microsecond)
3. Yield — 8 thread yields (~microseconds)
4. Sleep — 100us increments until deadline

When to use

• When dropping messages is unacceptable and brief blocking is acceptable
• Logging pipelines, recording, non-RT data transfer

When NOT to use

• Real-time control loops — blocking in tick() causes deadline misses
• Any node with .rate() or .budget() — use send() instead

Example

// simplified
use horus::prelude::*;

let topic = Topic::<DiagnosticReport>::new("diagnostics")?;

match topic.send_blocking(report, 10_u64.ms()) {
    Ok(()) => { /* sent */ }
    Err(SendBlockingError::Timeout) => {
        hlog!(warn, "Diagnostic buffer full for 10ms — dropping report");
    }
}

Receiving Methods

recv()

Takes the next unread message in FIFO order. Consumes it from the buffer.

Signature

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

Parameters

None.

Returns

• Some(T) — the oldest unread message (removed from buffer)
• None — no unread messages available

Behavior

• Non-blocking. Returns immediately.
• Each message is delivered to each subscriber exactly once. After recv() returns it, the message is consumed.
• Latency: recv() reads from the shared-memory ring in all cases. Same-core (publisher and subscriber on one executor thread) is L1-hot — ~19 ns one-way; a consumer on a different core, whether another thread or another process, is ~220 ns one-way (cache-coherence physics)

When to use

• Command streams where every message matters (velocity commands, goal updates)
• Processing pipelines where order matters (image frames, sensor sequences)

When NOT to use

• State-like data where you only care about the latest value — use read_latest() instead

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

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

// IMPORTANT: drain every tick to avoid stale data accumulation
while let Some(msg) = topic.recv() {
    process(msg);
}

try_recv()

Low-level non-blocking receive. Marked #[doc(hidden)] in the source — it is an internal/test entry point.

Signature

// simplified
pub fn try_recv(&self) -> Option<T>

Returns

Option<T> — same values as recv(), but try_recv() skips the same-thread fast path plus the logging/recording hooks that recv() runs. Prefer recv() in application code.

read_latest()

Reads the most recent message, skipping all older ones.

Signature

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

Parameters

None.

Returns

• Some(T) — copy of the newest message
• None — no messages available

Behavior

• Reads without consuming — multiple calls return the same value until a new message arrives
• Skips all older messages — only the latest matters
• Requires T: Copy because it reads (copies) without removing from the buffer

When to use

• State-like data where you only care about the current value: poses, sensor readings, configuration
• Slow subscribers that can't keep up with a fast publisher

When NOT to use

• Command streams where every message matters — use recv() instead
• Messages that aren't Copy (e.g., types with Vec, String)

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

let odom = Topic::<Odometry>::new("odom")?;

// Good: pose is state — only the latest matters
if let Some(pose) = odom.read_latest() {
    current_position = pose;
}

State & Metrics Methods

name()

Returns the topic name.

Signature

// simplified
pub fn name(&self) -> &str

Returns

&str — the name passed to new() or with_capacity().

has_message()

Checks whether at least one unread message exists.

Signature

// simplified
pub fn has_message(&self) -> bool

Returns

true if recv() would return Some. false if the buffer is empty.

pending_count()

Returns the number of unread messages in the buffer.

Signature

// simplified
pub fn pending_count(&self) -> u64

Returns

u64 — count of messages waiting to be consumed by recv().

dropped_count()

Returns the total number of messages dropped due to buffer-full overwrites.

Signature

// simplified
pub fn dropped_count(&self) -> u64

Returns

u64 — cumulative count since topic creation. Incremented by send() each time it drops the newly-sent message after the buffer stays full (it is the topic's send_failures counter).

When to use

• Monitor for backpressure — a rising dropped_count() means the subscriber is too slow
• Log warnings when drops exceed a threshold

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

let topic = Topic::<LaserScan>::new("scan")?;

if topic.dropped_count() > 0 {
    hlog!(warn, "Dropped {} scans — subscriber can't keep up", topic.dropped_count());
}

pub_count()

Returns the number of active publishers on this topic.

Advanced: These methods are #[doc(hidden)] in the Rust API — they may change without notice. For monitoring, prefer horus topic list --verbose.

Signature

// simplified
pub fn pub_count(&self) -> u32

sub_count()

Returns the number of active subscribers on this topic.

Advanced: These methods are #[doc(hidden)] in the Rust API — they may change without notice. For monitoring, prefer horus topic list --verbose.

Signature

// simplified
pub fn sub_count(&self) -> u32

is_same_process() — #[cfg(test)] only

Checks whether all publishers and subscribers are in the same process. Available only in test builds.

Signature

// simplified
#[cfg(test)]
pub fn is_same_process(&self) -> bool

Returns

true if all participants are in the same process. Note: every topic is SHM-backed even when same-process (so a cross-process consumer can join later), so being same-process does not change the backend — what actually moves latency is whether the participants share a core (see the send()/recv() latency notes).

Note: This method is gated behind #[cfg(test)] and is not available in production builds. It exists for internal testing and assertions only.

is_same_thread() — #[cfg(test)] only

Checks whether all publishers and subscribers are on the same thread. Available only in test builds.

Signature

// simplified
#[cfg(test)]
pub fn is_same_thread(&self) -> bool

Returns

true if all publishers and subscribers are on the same thread. (Every topic stays SHM-backed; same-thread participants share a core and hit the L1-hot fast path — ~12–19 ns one-way — but the backend is still shared memory.)

Note: This method is gated behind #[cfg(test)] and is not available in production builds. It exists for internal testing and assertions only.

metrics()

Returns aggregate send/receive statistics.

Signature

// simplified
pub fn metrics(&self) -> TopicMetrics

Returns

TopicMetrics — see Types section below.

Example

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

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

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

Pool-Backed Types (Zero-Copy)

For large data types (Image, PointCloud, DepthImage, Tensor), HORUS uses pool-backed allocation. The Topic API works the same — send(), recv(), try_send() — but the data is transferred via shared memory pools instead of copying through the ring buffer.

// simplified
use horus::prelude::*;

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

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

// IMPORTANT: call recv() every tick to drain — images are large and stale frames waste pool slots
if let Some(img) = camera.recv() {
    println!("{}x{} image received", img.width(), img.height());
}

See Image API, PointCloud API, DepthImage API, Tensor API for type-specific methods.

Types

TopicMetrics

Aggregate statistics returned by topic.metrics().

SendBlockingError

Error returned by send_blocking().

Production Example

Multi-sensor fusion node subscribing to two topics and publishing a fused pose:

// simplified
use horus::prelude::*;
use horus_robotics::prelude::*;

struct FusionNode {
    imu_sub: Topic<Imu>,
    odom_sub: Topic<Odometry>,
    pose_pub: Topic<Pose3D>,
    last_imu: Option<Imu>,
    last_odom: Option<Odometry>,
}

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

    fn tick(&mut self) {
        // IMPORTANT: always drain both topics every tick
        if let Some(imu) = self.imu_sub.recv() { self.last_imu = Some(imu); }
        if let Some(odom) = self.odom_sub.recv() { self.last_odom = Some(odom); }

        if let (Some(imu), Some(odom)) = (&self.last_imu, &self.last_odom) {
            let fused = self.fuse(imu, odom);
            self.pose_pub.send(fused);
        }
    }
}

See Also

• Topics Concept — Architecture, backends, and design patterns
• Topics for Beginners — Gentle introduction
• Communication Overview — When to use topics vs services vs actions
• Image API — Pool-backed camera images
• PointCloud API — Pool-backed 3D point clouds
• DepthImage API — Pool-backed depth images
• Tensor API — Tensor descriptor
• Scheduler API — Running nodes that use topics
• Python Topic API — Python topic bindings