Building Your Second Application

Now that you've built your first HORUS application, let's create something more practical: a 3-node sensor pipeline that reads temperature data, filters out noise, and displays the results.

What You'll Build

A real-time temperature monitoring system with:

  1. SensorNode: Publishes simulated temperature readings every second
  2. FilterNode: Subscribes to raw temperatures, filters noise, republishes clean data
  3. DisplayNode: Subscribes to filtered data, displays to console

This demonstrates:

  • Multi-node communication patterns
  • Data pipeline processing
  • Real-time filtering
  • Live inspection with horus monitor

Architecture

Loading diagram...
Temperature pipeline: SensorNode → FilterNode → DisplayNode

Step 1: Create the Project

horus new temperature_pipeline -r
cd temperature_pipeline

-r picks plain Rust without macros, which is what the listing below uses. Leave the flag off and horus new turns interactive instead, prompting for a language and then for macros - the equivalent answers are 2 (Rust) and N (no macros).

Step 2: Write the Code

Replace src/main.rs with this complete, runnable code:

use horus::prelude::*;
use std::time::{Duration, Instant};

// ============================================================================
// Node 1: SensorNode - Publishes temperature readings
// ============================================================================

struct SensorNode {
    temp_pub: Option<Topic<f32>>,
    last_publish: Instant,
    reading: f32,
}

impl SensorNode {
    fn new() -> Result<Self> {
        Ok(Self {
            temp_pub: None,
            last_publish: Instant::now(),
            reading: 20.0,
        })
    }
}

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

    fn init(&mut self) -> Result<()> {
        // Build topics here, not in new(): the scheduler only sets the node
        // context around init()/tick(), and Topic::new() uses it to record which
        // node owns the topic. Topics built in main() have no owner, so they
        // never show up under a node in `horus monitor`.
        self.temp_pub = Some(Topic::new("raw_temp")?);
        hlog!(info, "Temperature sensor initialized");
        Ok(())
    }

    fn tick(&mut self) {
        // Publish every 1 second
        if self.last_publish.elapsed() >= Duration::from_secs(1) {
            // Simulate realistic temperature with noise
            // Base temperature oscillates between 20-30°C
            let base_temp = 25.0 + (self.reading * 0.1).sin() * 5.0;

            // Add random noise (+/- 2°C)
            let noise = (self.reading * 0.7).sin() * 2.0;
            let temperature = base_temp + noise;

            // Publish raw temperature
            if let Some(ref temp_pub) = self.temp_pub {
                temp_pub.send(temperature);
            }

            hlog!(info, "Published raw temp: {:.2}°C", temperature);

            self.reading += 1.0;
            self.last_publish = Instant::now();
        }
    }

    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "Sensor shutdown complete");
        Ok(())
    }
}

// ============================================================================
// Node 2: FilterNode - Removes noise with exponential moving average
// ============================================================================

struct FilterNode {
    raw_sub: Option<Topic<f32>>,
    filtered_pub: Option<Topic<f32>>,
    filtered_value: Option<f32>,
    alpha: f32,  // Smoothing factor (0.0 - 1.0)
}

impl FilterNode {
    fn new() -> Result<Self> {
        Ok(Self {
            raw_sub: None,
            filtered_pub: None,
            filtered_value: None,
            alpha: 0.3,  // 30% new data, 70% previous (smooth but responsive)
        })
    }
}

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

    fn init(&mut self) -> Result<()> {
        self.raw_sub = Some(Topic::new("raw_temp")?);
        self.filtered_pub = Some(Topic::new("filtered_temp")?);
        hlog!(info, "Filter initialized (alpha = {:.2})", self.alpha);
        Ok(())
    }

    fn tick(&mut self) {
        // Check for new temperature reading
        if let Some(raw_temp) = self.raw_sub.as_ref().and_then(|s| s.recv()) {
            // Apply exponential moving average filter
            let filtered = match self.filtered_value {
                Some(prev) => self.alpha * raw_temp + (1.0 - self.alpha) * prev,
                None => raw_temp,  // First reading, no previous value
            };

            self.filtered_value = Some(filtered);

            // Publish filtered temperature
            if let Some(ref filtered_pub) = self.filtered_pub {
                filtered_pub.send(filtered);
            }

            hlog!(info, "Filtered: {:.2}°C -> {:.2}°C (removed {:.2}°C noise)",
                    raw_temp, filtered, raw_temp - filtered);
        }
    }

    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "Filter shutdown complete");
        Ok(())
    }
}

// ============================================================================
// Node 3: DisplayNode - Shows filtered temperature on console
// ============================================================================

struct DisplayNode {
    filtered_sub: Option<Topic<f32>>,
    display_counter: u32,
}

impl DisplayNode {
    fn new() -> Result<Self> {
        Ok(Self {
            filtered_sub: None,
            display_counter: 0,
        })
    }
}

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

    fn init(&mut self) -> Result<()> {
        self.filtered_sub = Some(Topic::new("filtered_temp")?);
        hlog!(info, "Display initialized");
        println!("\n========================================");
        println!("  Temperature Monitor - Press Ctrl+C to stop");
        println!("========================================\n");
        Ok(())
    }

    fn tick(&mut self) {
        if let Some(temp) = self.filtered_sub.as_ref().and_then(|s| s.recv()) {
            self.display_counter += 1;

            // Display temperature with visual indicator
            let status = if temp < 22.0 {
                "COLD"
            } else if temp > 28.0 {
                "HOT"
            } else {
                "NORMAL"
            };

            println!(
                "[Reading #{}] Temperature: {:.1}°C - Status: {}",
                self.display_counter, temp, status
            );

            hlog!(debug, "Displayed reading #{}", self.display_counter);
        }
    }

    fn shutdown(&mut self) -> Result<()> {
        println!("\n========================================");
        println!("  Total readings displayed: {}", self.display_counter);
        println!("========================================\n");
        hlog!(info, "Display shutdown complete");
        Ok(())
    }
}

// ============================================================================
// Main Application - Configure and run the scheduler
// ============================================================================

fn main() -> Result<()> {
    println!("Starting Temperature Pipeline...\n");

    let mut scheduler = Scheduler::new();

    // Add nodes in priority order:
    // 1. SensorNode (order 0) - Runs first to generate data
    scheduler.add(SensorNode::new()?).order(0).build()?;

    // 2. FilterNode (order 1) - Runs second to process data
    scheduler.add(FilterNode::new()?).order(1).build()?;

    // 3. DisplayNode (order 2) - Runs last to display results
    scheduler.add(DisplayNode::new()?).order(2).build()?;

    println!("All nodes registered. Running...\n");

    // Run the scheduler: it calls init() on every node, then ticks them
    // in .order() sequence (blocks until Ctrl+C)
    scheduler.run()?;

    Ok(())
}

Step 3: Run the Application

horus run

Expected Output:

Starting Temperature Pipeline...

All nodes registered. Running...

========================================
  Temperature Monitor - Press Ctrl+C to stop
========================================

[Reading #1] Temperature: 31.5°C - Status: HOT
[Reading #2] Temperature: 31.4°C - Status: HOT
[Reading #3] Temperature: 30.9°C - Status: HOT
[Reading #4] Temperature: 30.0°C - Status: HOT
[Reading #5] Temperature: 29.0°C - Status: HOT
[Reading #6] Temperature: 28.1°C - Status: HOT
[Reading #7] Temperature: 27.6°C - Status: NORMAL

The simulation is deterministic, so these values are exactly what you'll see. Only this application's own println! output is shown above. Your terminal will show more: hlog! writes to stderr, so [INFO] [SensorNode] Published raw temp: ... and [INFO] [FilterNode] Filtered: ... lines are interleaved, and the scheduler prints its own startup lines (Initialized node 'SensorNode', the dependency-graph summary) too.

Press Ctrl+C to stop:

^C
Ctrl+C received! Shutting down HORUS scheduler...

========================================
  Total readings displayed: 7
========================================

Step 4: Inspect the Running Pipeline

The monitor ships as a separate plugin, so install it once before using it:

horus install horus-monitor

Then start the pipeline again with horus run, open a second terminal, and run:

horus monitor

The monitor will show:

Nodes

  • SensorNode: Publishing to raw_temp every ~1 second
  • FilterNode: Subscribing to raw_temp, publishing to filtered_temp
  • DisplayNode: Subscribing to filtered_temp

Topics

  • raw_temp (f32): Noisy temperature readings
  • filtered_temp (f32): Smooth temperature readings

Metrics

  • IPC Latency: ~85ns p50 for these 1-publisher/1-subscriber topics (SpscShm backend; sub-microsecond!)
  • Tick Duration: How long each node takes to execute
  • Message Counts: Total messages sent/received

Understanding the Code

SensorNode

// Publish every 1 second
if self.last_publish.elapsed() >= Duration::from_secs(1) {
    let base_temp = 25.0 + (self.reading * 0.1).sin() * 5.0;
    let noise = (self.reading * 0.7).sin() * 2.0;
    let temperature = base_temp + noise;

    if let Some(ref temp_pub) = self.temp_pub {
        temp_pub.send(temperature);
    }
}

Key Points:

  • Uses Instant to track time between publishes
  • Simulates realistic sensor data with noise
  • Publishes to "raw_temp" topic

FilterNode

// Exponential moving average filter
let filtered = self.alpha * raw_temp + (1.0 - self.alpha) * prev;
if let Some(ref filtered_pub) = self.filtered_pub {
    filtered_pub.send(filtered);
}

Key Points:

  • Subscribes to "raw_temp", publishes to "filtered_temp"
  • Implements exponential moving average (EMA) filter
  • alpha = 0.3 balances responsiveness vs smoothness

Filter Behavior:

  • High alpha (0.8): Fast response, less smoothing
  • Low alpha (0.2): Slow response, more smoothing

DisplayNode

if let Some(temp) = self.filtered_sub.as_ref().and_then(|s| s.recv()) {
    println!("[Reading #{}] Temperature: {:.1}°C", count, temp);
}

Key Points:

  • Subscribes to "filtered_temp"
  • Only receives when new data is available
  • recv() returns None when no message (not an error!)

Common Issues & Fixes

Issue 1: No Output Displayed

Symptom:

Starting Temperature Pipeline...
All nodes registered. Running...

========================================
  Temperature Monitor - Press Ctrl+C to stop
========================================

[Nothing appears]

Cause: Topics not connecting (typo in topic names)

Fix:

  • Check topic names match exactly: "raw_temp" and "filtered_temp"
  • Verify with monitor: horus monitor -> Topics tab
  • Ensure all nodes are running in same scheduler

Issue 2: Too Much/Too Little Smoothing

Symptom: Temperature changes too fast or too slow

Fix: Adjust the alpha value in FilterNode:

alpha: 0.3,  // Current: moderate smoothing

// Try these alternatives:
alpha: 0.7,  // More responsive, less smooth
alpha: 0.1,  // Very smooth, slower response

Issue 3: Monitor Shows No Nodes

Symptom: Monitor is empty

Cause: Application not running or monitor started before app

Fix:

  1. Start the application first: horus run
  2. Then start monitor in separate terminal: horus monitor
  3. Monitor auto-discovers running nodes

Issue 4: Build Errors

Symptom:

error[E0433]: failed to resolve: use of undeclared type `Topic`

Fix:

  • Ensure HORUS is installed: horus --help
  • Check import: use horus::prelude::*;
  • Run from project directory (where horus.toml is)

Experiments to Try

1. Change Update Rate

Make the sensor publish faster:

// In SensorNode::tick()
if self.last_publish.elapsed() >= Duration::from_millis(500) {  // 2 Hz instead of 1 Hz

2. Add Temperature Alerts

Add to DisplayNode:

if temp > 30.0 {
    println!("  WARNING: High temperature detected!");
}

3. Log Data to File

Add to DisplayNode::tick():

use std::fs::OpenOptions;
use std::io::Write;

let mut file = OpenOptions::new()
    .create(true)
    .append(true)
    .open("temperature_log.txt")
    .unwrap();

writeln!(file, "{:.1}", temp).ok();

4. Add Multiple Sensors

Create a second sensor node. Node names are the identity key for the watchdog, the shared-memory registry slot and the pause/kill control flags, so the scheduler refuses duplicate names - run() returns an error before the first tick. Give SensorNode a name of its own first:

// 1. Make the name configurable
struct SensorNode {
    name: String,
    temp_pub: Option<Topic<f32>>,
    last_publish: Instant,
    reading: f32,
}

impl SensorNode {
    fn new(name: &str) -> Result<Self> {
        Ok(Self {
            name: name.to_string(),
            temp_pub: None,
            last_publish: Instant::now(),
            reading: 20.0,
        })
    }
}

impl Node for SensorNode {
    fn name(&self) -> &str {
        &self.name
    }

    // init(), tick() and shutdown() are unchanged
}

// 2. In main(), replace the single `scheduler.add(SensorNode::new()?)` line
//    with two sensors that have distinct names
scheduler.add(SensorNode::new("Sensor1")?).order(0).build()?;
scheduler.add(SensorNode::new("Sensor2")?).order(0).build()?;

Both will publish to the same topic, and FilterNode will process both!

Key Concepts Demonstrated

Pipeline Pattern: Data flows through stages (Sensor -> Filter -> Display)

Pub/Sub Decoupling: Nodes don't know about each other, only topics

Real-Time Processing: Filtering happens as data arrives

Shared Memory IPC: Sub-microsecond communication between nodes

Priority Scheduling: Sensor runs before filter, filter before display

Next Steps

Now that you've built a 3-node pipeline, try:

  1. Testing - Learn how to unit test your nodes
  2. Using Pre-Built Nodes - Use library nodes instead of writing from scratch
  3. node! Macro - Reduce boilerplate with macros
  4. Message Types - Use complex message types instead of primitives

Putting It Together

There is no second listing here - the code in Step 2 above is the whole program, and it is complete and runnable as-is. It is still a teaching example, though: the "sensor" is a sine wave and the display is a println!, so swap both for real hardware and a real sink before you rely on it. To save it:

  1. Copy the entire code block from Step 2
  2. Replace src/main.rs in your project
  3. Run with horus run

For additional examples, see Basic Examples.