Struct Receiver 

pub struct Receiver<T> { /* private fields */ }

Implementations

impl<T> Receiver<T>

pub fn map<U, F>(self, f: F) -> Receiver<U>

where F: FnMut(T) -> U + Send + 'static, T: Send + 'static, U: Send + 'static,

Map (transform) messages as they are received

Creates a new receiver that applies a transformation function to each message. This is useful for data transformations like resizing images, converting units, or extracting fields from complex messages.

Arguments

• f - Transformation function to apply to each message

Example

// Resize camera images on the fly
let mut resized_images = ctx.subscribe(sensor::CAMERA_RGB)
    .await?
    .map(|img| resize_image(img, 320, 240));

while let Some(small_image) = resized_images.recv().await {
    // Process resized image
}

// Extract specific fields from complex messages
let mut speeds = ctx.subscribe(actuator::CMD_VEL)
    .await?
    .map(|twist| twist.linear[0]);

while let Some(speed) = speeds.recv().await {
    println!("Current speed: {}", speed);
}

pub fn filter<F>(self, predicate: F) -> Receiver<T>

where F: FnMut(&T) -> bool + Send + 'static, T: Send + 'static,

Filter messages based on a predicate

Creates a new receiver that only passes through messages that satisfy the predicate. This is useful for filtering sensor data, ignoring invalid readings, or implementing conditional processing.

Arguments

• predicate - Function that returns true for messages to keep, false to drop

Example

// Only process high-speed motor commands
let mut high_speeds = ctx.subscribe(actuator::CMD_VEL)
    .await?
    .filter(|twist| twist.linear[0] > 0.5);

while let Some(twist) = high_speeds.recv().await {
    println!("High speed detected: {}", twist.linear[0]);
}

// Filter out invalid lidar readings
let mut valid_scans = ctx.subscribe(sensor::LIDAR_SCAN)
    .await?
    .filter(|scan| scan.ranges.iter().all(|&r| r > 0.0 && r < 100.0));

while let Some(scan) = valid_scans.recv().await {
    // All ranges are valid
}

pub fn filter_map<U, F>(self, f: F) -> Receiver<U>

where F: FnMut(T) -> Option<U> + Send + 'static, T: Send + 'static, U: Send + 'static,

Filter and map messages in one operation

Combines filtering and mapping into a single efficient operation. The function returns Some(U) for messages to keep (and transform), or None to drop. This is more efficient than chaining .filter() and .map() separately.

Arguments

• f - Function that returns Some(transformed) to keep, None to drop

Example

// Extract speed only when robot is moving
let mut active_speeds = ctx.subscribe(actuator::CMD_VEL)
    .await?
    .filter_map(|twist| {
        if twist.linear[0] > 0.1 {
            Some(twist.linear[0])
        } else {
            None
        }
    });

while let Some(speed) = active_speeds.recv().await {
    println!("Robot moving at: {}", speed);
}

// Parse valid diagnostics readings
let mut parsed_data = ctx.subscribe(system::DIAGNOSTICS)
    .await?
    .filter_map(|raw| parse_sensor_data(&raw).ok());

while let Some(data) = parsed_data.recv().await {
    // Process successfully parsed data
}

impl<T> Receiver<T>

pub fn debounce(self, duration: Duration) -> Receiver<T>

where T: Send + 'static,

Debounce (rate-limit) messages to prevent rapid bursts

Creates a new receiver that only emits a message if a certain duration has passed since the last emission. This is useful for reducing update frequency of high-rate sensors or preventing UI update storms.

Strategy: “Emit latest after quiet period”

• Messages are buffered during the debounce period
• Only the most recent message is emitted after the period expires
• Earlier messages within the period are dropped

Arguments

• duration - Minimum time between emissions

Example

// Reduce camera updates to max 10 Hz (100ms debounce)
let mut debounced_images = ctx.subscribe(sensor::CAMERA_RGB)
    .await?
    .debounce(Duration::from_millis(100));

while let Some(image) = debounced_images.recv().await {
    // Process at most 10 images/sec
}

// Prevent rapid velocity command updates
let mut stable_status = ctx.subscribe(actuator::CMD_VEL)
    .await?
    .debounce(Duration::from_millis(50));

while let Some(status) = stable_status.recv().await {
    // Only get updates every 50ms
}

pub fn window(self, window_duration: Duration) -> Receiver<Vec<T>>

where T: Send + 'static,

Window messages over a time period

Creates a new receiver that collects messages over a fixed time window and emits them as a batch (Vec) at the end of each window. This is useful for:

• Aggregating sensor readings (averaging, min/max)
• Batch processing for efficiency
• Time-series analysis
• Rate calculation (count messages per window)

Behavior

• Fixed Windows: Each window has a fixed duration
• Non-overlapping: Windows don’t overlap (tumbling window)
• Batch Emission: All messages in window emitted as Vec at window end
• Empty Windows: Empty Vec emitted if no messages received
• Termination: Emits final partial window when source ends

Arguments

• window_duration - Duration of each time window

Example

// Average IMU readings over 100ms windows
let mut windowed = ctx.subscribe(sensor::IMU)
    .await?
    .window(Duration::from_millis(100));

while let Some(batch) = windowed.recv().await {
    if !batch.is_empty() {
        let avg = batch.iter().map(|imu| imu.linear_acceleration[0]).sum::<f32>() / batch.len() as f32;
        println!("Average accel: {} (n={})", avg, batch.len());
    }
}

// Count messages per second
let mut counts = ctx.subscribe(sensor::BATTERY_VOLTAGE)
    .await?
    .window(Duration::from_secs(1))
    .map(|batch| batch.len());

while let Some(count) = counts.recv().await {
    println!("Events/sec: {}", count);
}

// Batch process for efficiency
let mut batches = ctx.subscribe(sensor::CAMERA_RGB)
    .await?
    .window(Duration::from_millis(50));

while let Some(batch) = batches.recv().await {
    // Process entire batch at once (more efficient than one-by-one)
    process_batch(&batch).await?;
}

Performance

• Memory: Buffers all messages within window (use reasonable window sizes)
• Latency: Adds latency equal to window duration
• CPU: Minimal overhead for window management

Caveats

• Window size affects memory usage (smaller windows = less memory)
• All messages are delivered at window end (batch latency)
• For sliding windows, consider multiple overlapping .window() subscriptions

impl<T> Receiver<T>

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

Receive the next message

Returns None when all senders have been dropped.

pub fn try_recv(&mut self) -> Result<T, TryRecvError>

Try to receive a message without blocking

Returns an error if the channel is empty or all senders have been dropped.

pub fn is_empty(&self) -> bool

Check if the channel is currently empty

Note: This is a point-in-time check and may change immediately after.

pub fn len(&self) -> usize

Get the number of messages currently in the channel

This is useful for monitoring backpressure and queue depth.

Example

let depth = rx.len();
if depth > 50 {
    warn!("Channel backlog growing: {} messages", depth);
}

pub fn with_capacity(self, capacity: usize) -> Self

where T: 'static + Send,

Convert an unbounded receiver to a bounded one with the specified capacity

This creates a new bounded channel and spawns a task to forward messages.

Arguments

• capacity - Maximum number of messages to buffer

Backpressure Strategy

When the bounded channel is full, forwarding blocks (waits) until the consumer makes room. No messages are dropped - this is the “bound without dropping” strategy. If you want old messages evicted instead of blocking the publisher, use Receiver::with_drop_oldest.

Example

let mut images = ctx.subscribe(sensor::CAMERA_RGB)
    .await?
    .with_capacity(50);  // Limit to 50 images in queue

// If publisher sends faster than we process, old images are dropped
while let Some(image) = images.recv().await {
    // Always get relatively recent images
}

pub fn with_drop_oldest(self, capacity: usize) -> Self

where T: 'static + Send,

Convert an unbounded receiver to a bounded one that drops oldest messages

Similar to with_capacity(), but explicitly uses a “drop oldest” strategy when the buffer is full. This is useful for real-time data where the most recent value is more important than historical data.

Arguments

• capacity - Maximum number of messages to buffer

Example

// For sensor data, we want the latest reading
let mut lidar = ctx.subscribe(sensor::LIDAR_SCAN)
    .await?
    .with_drop_oldest(10);  // Keep only 10 most recent scans

while let Some(scan) = lidar.recv().await {
    // Always process recent data
}

pub fn capacity(&self) -> Option<usize>

Get a reference to the maximum capacity if this is a bounded channel

Returns None for unbounded channels or the capacity for bounded ones.

Example

if let Some(cap) = rx.capacity() {
    println!("Channel capacity: {}", cap);
    println!("Current usage: {}/{}", rx.len(), cap);
    println!("Utilization: {:.1}%", (rx.len() as f32 / cap as f32) * 100.0);
} else {
    println!("Unbounded channel, {} messages queued", rx.len());
}

pub fn utilization(&self) -> Option<f32>

Get the current utilization ratio (0.0 to 1.0) for bounded channels

Returns None for unbounded channels.

Example

if let Some(utilization) = rx.utilization() {
    if utilization > 0.8 {
        warn!("Channel is {}% full, backpressure may occur", utilization * 100.0);
    }
}

impl<T> Receiver<T>

pub fn zip<U>(self, other: Receiver<U>) -> Receiver<(T, U)>

where T: Send + 'static, U: Send + 'static,

Combine two streams by pairing their messages

Creates a new stream that emits pairs of messages from two input streams. The combined stream emits (T, U) tuples when both streams have messages available.

Behavior

• Paired Emission: Waits for both streams to have messages, then emits the pair
• Buffering: Buffers messages from faster stream until slower stream catches up
• Termination: Stops when either stream ends
• Capacity: Uses the smaller capacity of the two input streams

Use Cases

• Combining sensor readings (e.g., camera + LiDAR)
• Synchronizing multiple data sources
• Correlating events from different streams
• Sensor fusion

Example

use mecha10_core::prelude::*;
use mecha10_core::messages::{Image, LaserScan};
use mecha10_core::topics::sensor;

// Subscribe to two sensor streams
let camera = ctx.subscribe(sensor::CAMERA_RGB).await?;
let lidar = ctx.subscribe(sensor::LIDAR_SCAN).await?;

// Combine them
let mut fused = camera.zip(lidar);

// Process paired messages
while let Some((img, scan)) = fused.recv().await {
    // Both sensors have data - process together
    process_sensor_fusion(&img, &scan).await?;
}

Performance

• Memory: Buffers messages from faster stream (bounded by capacity)
• Latency: Adds latency equal to slower stream’s inter-message delay
• CPU: Minimal overhead for pairing logic

Caveats

• If streams have very different rates, slower stream will be the bottleneck
• Buffer can fill up if rate difference is sustained
• Consider .debounce() on faster stream if rate mismatch is severe

Trait Implementations

impl<T> From<Receiver<T>> for Receiver<T>

fn from(inner: Receiver<T>) -> Self

Converts to this type from the input type.

impl<T> From<UnboundedReceiver<T>> for Receiver<T>

fn from(inner: UnboundedReceiver<T>) -> Self

Converts to this type from the input type.