Struct MessageBus 

pub struct MessageBus { /* private fields */ }

Message bus for pub/sub communication

Implementations

impl MessageBus

pub async fn connect(redis_url: &str, node_id: &str) -> Result<Self>

Connect to Redis and create a new message bus

Uses exponential backoff retry logic for resilient connection establishment.

Arguments

• redis_url - Redis connection URL (e.g., “redis://localhost:6379”)
• node_id - Unique identifier for this node

Example

let bus = MessageBus::connect("redis://localhost:6379", "camera-node").await?;

pub async fn connect_with_retry( redis_url: &str, node_id: &str, retry_config: RetryConfig, ) -> Result<Self>

Connect to Redis with custom retry configuration

Arguments

• redis_url - Redis connection URL
• node_id - Unique identifier for this node
• retry_config - Retry configuration for exponential backoff

Example

let retry_config = RetryConfig {
    max_attempts: 10,
    initial_delay: Duration::from_millis(200),
    max_delay: Duration::from_secs(30),
    multiplier: 2.0,
};
let bus = MessageBus::connect_with_retry(
    "redis://localhost:6379",
    "camera-node",
    retry_config
).await?;

pub fn set_namespace(&mut self, namespace: &str)

Set the namespace for topic isolation

Namespaces allow multiple robots/fleets to use the same Redis instance without topic collisions.

pub async fn publish<T: Serialize>( &mut self, topic: &str, payload: &T, ) -> Result<()>

Publish a message to a topic

Arguments

• topic - Topic name (e.g., “/scan”, “/odom”)
• payload - Message payload (must be serializable)

Example

bus.publish("/position", &Point { x: 1.0, y: 2.0 }).await?;

pub async fn subscribe<T: DeserializeOwned + Send + 'static>( &mut self, topic: &str, consumer_group: &str, ) -> Result<Subscriber<T>>

Subscribe to a topic with a consumer group

Consumer groups allow multiple nodes to process messages in parallel, with each message delivered to only one consumer in the group.

Arguments

• topic - Topic name (e.g., “/scan”, “/odom”)
• consumer_group - Consumer group name (e.g., “processing”, “logging”)

Example

let mut rx = bus.subscribe::<LaserScan>("/scan", "slam").await?;
while let Some(msg) = rx.recv().await {
    println!("Received: {:?}", msg.payload);
    msg.ack().await?;
}

pub async fn subscribe_pattern<T: DeserializeOwned + Send + 'static>( &mut self, pattern: &str, consumer_group: &str, ) -> Result<Subscriber<T>>

Subscribe to multiple topics matching a wildcard pattern

Enables aggregated subscriptions across multiple topics using wildcard patterns. Useful for collecting data from multiple robots (e.g., */camera/rgb subscribes to camera feeds from all robots).

Arguments

• pattern - Wildcard pattern with * as placeholder (e.g., “/camera/rgb”, “robot-/scan”)
• consumer_group - Consumer group name for load balancing

Pattern Syntax

• * matches any segment in the topic path
• */camera/rgb matches “robot-1/camera/rgb”, “robot-2/camera/rgb”, etc.
• robot-*/scan matches “robot-1/scan”, “robot-alpha/scan”, etc.

Example

// Subscribe to camera feeds from ALL robots
let mut rx = bus.subscribe_pattern::<Frame>("*/camera/rgb", "vision-processing").await?;

while let Some(msg) = rx.recv().await {
    println!("Received frame from robot: {}", msg.topic);
    // Process frame from any robot
    msg.ack().await?;
}

Implementation

This method:

1. Converts wildcard pattern to Redis KEYS pattern
2. Discovers all matching topics using discover_topics()
3. Subscribes to each discovered topic with the same consumer group
4. Multiplexes messages from all topics into a single receiver

Note

This performs initial topic discovery using Redis KEYS command. For dynamic topic discovery (new topics appearing after subscription), consider polling discover_topics() periodically and creating new subscriptions.

pub async fn discover_topics(&self, pattern: &str) -> Result<Vec<String>>

Discover topics matching a Redis KEYS pattern

Uses Redis KEYS command to find all streams matching the given pattern. This is useful for aggregated topic subscriptions.

Arguments

• pattern - Redis KEYS pattern (e.g., “mecha10:topic:/sensor/camera/*”)

Returns

A vector of topic paths (without the namespace prefix)

Example

// Discover all camera topics
let topics = bus.discover_topics("mecha10:topic:/sensor/camera/*").await?;
println!("Found {} camera topics", topics.len());

Note

This uses Redis KEYS command which can be slow on large datasets. For production use with many topics, consider using SCAN instead.

pub async fn close(&mut self) -> Result<()>

Unsubscribe from all topics and close connections