Topic and Pub/Sub

// 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();

use horus::prelude::*;

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

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);
    }
}

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);
        }
    }
}

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

pub fn send(&self, msg: T)

// send() always succeeds — no error handling needed
topic.send(data);

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

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

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

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

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

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

use std::time::Duration;

// Block up to 1ms waiting for buffer space
match topic.send_blocking(emergency_stop_cmd, Duration::from_millis(1)) {
    Ok(()) => { /* delivered */ }
    Err(_) => { hlog!(error, "Failed to deliver emergency stop!"); }
}

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

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

pub fn has_message(&self) -> bool

pub fn pending_count(&self) -> u64

// Topic name
let name = topic.name();  // "velocity"

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

// Message loss tracking
let dropped = topic.dropped_count();
if dropped > 0 {
    hlog!(warn, "Dropped {} messages on '{}'", dropped, topic.name());
}

// Connection monitoring
println!("Publishers: {}, Subscribers: {}", topic.pub_count(), topic.sub_count());

// Backend info (auto-selected by HORUS)
println!("Using backend: {}", topic.backend_name());

// Detailed metrics
let metrics = topic.metrics();
println!("Sent: {}, Received: {}", metrics.messages_sent(), metrics.messages_received());
println!("Send failures: {}, Recv failures: {}", metrics.send_failures(), metrics.recv_failures());

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

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

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

use horus::prelude::*;

// Define topics with compile-time type checking
topics! {
    pub CMD_VEL: CmdVel = "cmd_vel",
    pub SENSOR_DATA: SensorReading = "sensor.data",
    pub MOTOR_STATUS: MotorState = "motor.status",
}

// Publisher — type is enforced at compile time
let pub_topic: Topic<CmdVel> = Topic::new(CMD_VEL.name())?;
pub_topic.send(CmdVel::new(1.0, 0.5));

// Subscriber — same type enforced
let sub_topic: Topic<CmdVel> = Topic::new(CMD_VEL.name())?;
if let Some(cmd) = sub_topic.recv() {
    // cmd is guaranteed to be CmdVel
}

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 publisher
struct Broadcaster {
    broadcast_pub: Topic<String>,
}

// Multiple subscribers — all receive the same messages
struct Listener1 { broadcast_sub: Topic<String> }
struct Listener2 { broadcast_sub: Topic<String> }
struct Listener3 { broadcast_sub: Topic<String> }

// 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);
        }
    }
}

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

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

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

// WRONG - Do NOT use slashes
let topic = Topic::new("sensors/lidar");   // Will cause errors!
let topic = Topic::new("robot/cmd_vel");   // Will cause errors!

// 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

// send() always succeeds — no error handling needed
topic.send(data);

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

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

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")?;

use serde::{Serialize, Deserialize};

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

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

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

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);
        }
    }
}

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

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);
        }
    }
}

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);
            }
        }
    }
}

# 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

horus clean --shm            # Clean up

// 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")?;

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

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

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

self.pub_topic.send(data);