Struct Context 

pub struct Context { /* private fields */ }

Implementations

impl Context

pub fn builder(node_id: impl Into<String>) -> ContextBuilder

Create a context builder

Shorthand for ContextBuilder::new(node_id).

Example

let ctx = Context::builder("my-node")
    .with_instance("instance-1")
    .build()
    .await?;

impl Context

pub async fn new(node_id: &str) -> Result<Self>

Create a new context

Arguments

• node_id - Unique identifier for this node

Example

let ctx = Context::new("my-node").await?;

pub fn with_instance(self, instance: &str) -> Self

Set an instance name for this context (builder pattern)

This allows the same node implementation to run multiple instances with different names. The node_id becomes “{node_name}/{instance}”.

Example

// Single instance (no instance name)
let ctx = Context::new("camera_fake").await?;
// node_id: "camera_fake"

// Multi-instance (with instance names)
let left_ctx = Context::new("camera_fake").await?.with_instance("left");
// node_id: "camera_fake/left"

let right_ctx = Context::new("camera_fake").await?.with_instance("right");
// node_id: "camera_fake/right"

pub fn instance(&self) -> Option<&str>

Get the instance name if set

Returns None for single-instance nodes, or Some(instance) for multi-instance nodes.

pub fn get_redis_url() -> Result<String>

Get Redis URL from infrastructure configuration

pub fn node_id(&self) -> &str

Get the node ID

pub async fn load_node_config<T>(&self, node_name: &str) -> Result<T>

where T: DeserializeOwned + Default,

Load node configuration with environment-aware file-level fallback

This method implements the V2 configuration system with:

• Environment-specific configs (dev/staging/production)
• File-level fallback (no key-by-key merging)
• Automatic path resolution

Fallback Priority:

1. configs/{env}/nodes/{node-name}/config.json (environment-specific)
2. configs/common/nodes/{node-name}/config.json (common fallback)
3. T::default() (code default)

Environment Detection:

• Uses MECHA10_ENVIRONMENT env var (default: “dev”)
• Supported values: “dev”, “staging”, “production”, or custom

Type Parameters

• T - Configuration type that implements DeserializeOwned + Default

Arguments

• node_name - Name of the node (e.g., “simulation-bridge”, “camera-streamer”)

Returns

The loaded configuration of type T, using the first available source in the fallback chain.

Example

use mecha10::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Deserialize, Serialize, Default)]
struct SimBridgeConfig {
    camera_width: u32,
    camera_height: u32,
    camera_fps: u32,
}

// Loads from configs/{env}/nodes/simulation-bridge/config.json
// Falls back to configs/common/nodes/simulation-bridge/config.json
// Falls back to SimBridgeConfig::default()
let config: SimBridgeConfig = ctx.load_node_config("simulation-bridge").await?;

pub async fn has_control_plane(&self) -> bool

Check if control plane Redis connection is available

pub fn get_control_plane_redis_url() -> Result<Option<String>>

Get control plane Redis URL from infrastructure configuration

pub async fn subscribe<T>(&self, topic: Topic<T>) -> Result<Receiver<T>>

where T: Message + DeserializeOwned + Send + 'static,

Subscribe to a topic with type-safe message handling

Arguments

• topic - Type-safe topic to subscribe to

Returns

A receiver that will yield messages of type T

Example

use mecha10::prelude::*;
use mecha10::topics::sensor;

// Type is inferred as Receiver<Image>
let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;

while let Some(image) = images.recv().await {
    println!("Image: {}x{}", image.width, image.height);
}

pub async fn publish_to<T>(&self, topic: Topic<T>, message: &T) -> Result<()>

where T: Message + Serialize,

Publish a message to a specific topic

Arguments

• topic - Type-safe topic to publish to
• message - Message to publish (type checked against topic)

Example

use mecha10::prelude::*;
use mecha10::topics::sensor;
use mecha10::messages::Image;

let image = Image {
    timestamp: now_micros(),
    width: 640,
    height: 480,
    encoding: "rgb8".to_string(),
    data: vec![],
};

// Type checked: image must be of type Image
ctx.publish_to(sensor::CAMERA_RGB, &image).await?;

pub async fn publish_to_path<T>( &self, topic_path: &str, message: &T, ) -> Result<()>

where T: Message + Serialize,

Publish a message to a dynamic topic path (for runtime-generated topics like camera streams)

This is useful when the topic path is determined at runtime and cannot be a &'static str.

Arguments

• topic_path - The topic path as a String (e.g., “robot/sensors/camera/front/compressed”)
• message - The message to publish

pub async fn publish_to_scoped<T>( &self, topic: Topic<T>, message: &T, ) -> Result<()>

where T: Message + Serialize,

Publish a message to an instance-scoped topic

This automatically appends the context’s instance name to the topic path. If no instance is set, this behaves like publish_to().

Arguments

• topic - Type-safe topic to publish to
• message - Message to publish (type checked against topic)

Example

use mecha10::prelude::*;
use mecha10::topics::sensor;
use mecha10::messages::Image;

let ctx = Context::new("camera_fake").await?.with_instance("left");

let image = Image {
    timestamp: now_micros(),
    width: 640,
    height: 480,
    encoding: "rgb8".to_string(),
    data: vec![],
};

// Publishes to "/sensor/camera/rgb/left"
ctx.publish_to_scoped(sensor::CAMERA_RGB, &image).await?;

pub async fn publish_to_instance<T>( &self, topic: Topic<T>, instance: &str, message: &T, ) -> Result<()>

where T: Message + Serialize,

Publish a message to a topic with an explicit instance suffix

This allows publishing to a specific instance’s topic regardless of the context’s own instance name.

Arguments

• topic - Type-safe topic to publish to
• instance - Instance name to append to topic path
• message - Message to publish (type checked against topic)

Example

use mecha10::prelude::*;
use mecha10::topics::actuator;

let cmd = MotorCommand { velocity: 1.0, torque: 0.0 };

// Send to specific motor instances
ctx.publish_to_instance(actuator::MOTOR_CMD, "front_left", &cmd).await?;
ctx.publish_to_instance(actuator::MOTOR_CMD, "front_right", &cmd).await?;

pub async fn subscribe_scoped<T>( &self, topic: Topic<T>, instance: &str, ) -> Result<Receiver<T>>

where T: Message + DeserializeOwned + Send + 'static,

Subscribe to an instance-scoped topic

This automatically appends the context’s instance name to the topic path. If no instance is set, this behaves like subscribe().

Example

use mecha10::prelude::*;
use mecha10::topics::sensor;

let ctx = Context::new("vision_node").await?;

// Subscribe to left camera (topic: "/sensor/camera/rgb/left")
let mut left_images = ctx.subscribe_scoped(sensor::CAMERA_RGB, "left").await?;

// Subscribe to right camera (topic: "/sensor/camera/rgb/right")
let mut right_images = ctx.subscribe_scoped(sensor::CAMERA_RGB, "right").await?;

pub async fn subscribe_raw<T>(&self, topic_path: &str) -> Result<Receiver<T>>

where T: Message + DeserializeOwned + Send + 'static,

Subscribe to a topic using a raw string path (for RPC and advanced use cases)

Arguments

• topic_path - Raw topic path string

Returns

A receiver that will yield messages of type T

Note

This is a low-level method. Prefer using subscribe() with type-safe topics for normal use cases. This method is primarily for RPC and dynamic topic handling.

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

Discover topics matching a glob pattern

Queries Redis for all topics matching the given pattern without subscribing. Useful for inspecting available topics before subscribing.

Pattern Syntax

• * matches any sequence of characters within a path segment
• ** matches any sequence of path segments
• {a,b} matches either a or b (future enhancement)

Arguments

• pattern - Glob pattern to match topics against

Returns

A vector of topic paths that match the pattern

Example

use mecha10::prelude::*;

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

pub async fn subscribe_aggregated<T>( &self, pattern: &str, ) -> Result<AggregatedReceiver<T>>

where T: Message + DeserializeOwned + Send + 'static,

Subscribe to multiple topics matching a glob pattern

Returns an aggregated receiver that multiplexes all matching topics into a single stream. Each message is tagged with its source topic path.

Pattern Syntax

• * matches any sequence of characters within a path segment
• ** matches any sequence of path segments
• {a,b} matches either a or b (future enhancement)

Arguments

• pattern - Glob pattern to match topics against

Returns

An AggregatedReceiver<T> that yields (topic_path, message) tuples

Examples

use mecha10::prelude::*;

// Subscribe to all cameras with one call
let mut cameras = ctx.subscribe_aggregated::<ImageMessage>("/sensor/camera/*").await?;

while let Some((topic, image)) = cameras.recv().await {
    let camera_name = topic.rsplit('/').next().unwrap_or("unknown");
    println!("Camera {}: {}x{}", camera_name, image.width, image.height);
}

pub async fn publish_raw<T>(&self, topic_path: &str, message: &T) -> Result<()>

where T: Message + Serialize,

Publish a message to a topic using a raw string path (for RPC and advanced use cases)

Arguments

• topic_path - Raw topic path string
• message - Message to publish

Note

This is a low-level method. Prefer using publish_to() with type-safe topics for normal use cases. This method is primarily for RPC and dynamic topic handling.

pub async fn is_shutdown(&self) -> bool

Check if shutdown has been requested

pub async fn shutdown(&self)

Request shutdown

pub async fn wait_for_shutdown(&self)

Wait for shutdown signal

pub fn logging_span(&self) -> Span

Enter a logging span with this node’s context

This returns a guard that, when entered, adds the node_id to all log messages. The span is automatically exited when the guard is dropped.

Example

use mecha10::prelude::*;

let _guard = ctx.logging_span().entered();
info!("This message includes node_id automatically");
debug!("So does this one");

pub async fn subscribe_all_cameras(&self) -> Result<AggregatedReceiver<Image>>

Subscribe to all camera topics

Convenience method for subscribe_aggregated::<Image>("/sensor/camera/*")

Example

use mecha10::prelude::*;

let mut cameras = ctx.subscribe_all_cameras().await?;
while let Some((topic, image)) = cameras.recv().await {
    let camera_name = topic.rsplit('/').next().unwrap_or("unknown");
    println!("Camera {}: {}x{}", camera_name, image.width, image.height);
}

pub async fn discover_all_sensors(&self) -> Result<Vec<String>>

Subscribe to all sensor topics

Subscribes to all sensor data including cameras, LiDAR, IMU, odometry, etc.

Example

use mecha10::prelude::*;

// Subscribe to all sensors (returns untyped messages)
let topics = ctx.discover_topics("/sensor/**").await?;
println!("Found {} sensor topics", topics.len());

pub async fn subscribe_all_lidars( &self, ) -> Result<AggregatedReceiver<LaserScan>>

Subscribe to all LiDAR topics

Convenience method for subscribe_aggregated::<LaserScan>("/sensor/lidar/*")

pub async fn subscribe_all_imus(&self) -> Result<AggregatedReceiver<IMU>>

Subscribe to all IMU topics

Convenience method for subscribe_aggregated::<IMU>("/sensor/imu/*")

pub async fn subscribe_all_motors( &self, ) -> Result<AggregatedReceiver<MotorStatus>>

Subscribe to all motor topics

Convenience method for subscribe_aggregated::<MotorStatus>("/actuator/motor/*")

pub async fn subscribe_pattern<T>( &self, topic: Topic<T>, pattern: &str, ) -> Result<AggregatedReceiver<T>>

where T: Message + DeserializeOwned + Send + 'static,

Subscribe to a topic pattern with wildcard

Generic helper for subscribing to multiple topics matching a pattern.

Example

use mecha10::prelude::*;
use mecha10::topics::sensor;

// Subscribe to all instances of a specific topic type
let mut cameras = ctx.subscribe_pattern(sensor::CAMERA_RGB, "*").await?;

pub async fn state_manager(&self) -> Result<Arc<ConcreteStateManager>>

Get or create a state manager for this node

The state manager provides persistent storage for node state across restarts. By default, it uses filesystem storage in ./state/ directory.

Example

use mecha10::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
struct MyState {
    counter: u64,
    position: (f64, f64),
}

let state_mgr = ctx.state_manager().await?;

// Load previous state
let mut state = state_mgr
    .load::<MyState>("my_node")
    .await?
    .unwrap_or(MyState {
        counter: 0,
        position: (0.0, 0.0),
    });

// Update state
state.counter += 1;

// Save state
state_mgr.save("my_node", &state).await?;

pub async fn set_state_manager(&self, manager: Arc<ConcreteStateManager>)

Set a custom state manager for this context

This allows using alternative backends (Redis, PostgreSQL, etc.) or custom implementations.

Example

use mecha10::prelude::*;
use std::sync::Arc;

// Use in-memory state manager for testing
let mem_mgr = Arc::new(MemoryStateManager::new());
ctx.set_state_manager(mem_mgr).await;

pub async fn list_sensor_topics(&self) -> Result<Vec<String>>

List all sensor topics currently active in the system

Returns all topics under /sensor/* including cameras, LiDAR, IMU, GPS, etc.

Example

use mecha10::prelude::*;

let sensors = ctx.list_sensor_topics().await?;
println!("Available sensors:");
for topic in sensors {
    println!("  - {}", topic);
}

pub async fn list_actuator_topics(&self) -> Result<Vec<String>>

List all actuator topics currently active in the system

Returns all topics under /actuator/* including motors, servos, grippers, etc.

Example

use mecha10::prelude::*;

let actuators = ctx.list_actuator_topics().await?;
println!("Available actuators:");
for topic in actuators {
    println!("  - {}", topic);
}

pub async fn list_all_topics(&self) -> Result<Vec<String>>

List all topics in the system

Returns all active topics regardless of category.

Example

use mecha10::prelude::*;

let all_topics = ctx.list_all_topics().await?;
println!("All active topics ({} total):", all_topics.len());
for topic in all_topics {
    println!("  - {}", topic);
}

pub async fn list_camera_topics(&self) -> Result<Vec<String>>

List all camera topics

Returns all topics under /sensor/camera/*.

Example

use mecha10::prelude::*;

let cameras = ctx.list_camera_topics().await?;
println!("Available cameras:");
for topic in cameras {
    println!("  - {}", topic);
}

pub async fn list_lidar_topics(&self) -> Result<Vec<String>>

List all LiDAR topics

Returns all topics under /sensor/lidar/*.

pub async fn list_imu_topics(&self) -> Result<Vec<String>>

List all IMU topics

Returns all topics under /sensor/imu/*.

pub async fn list_motor_topics(&self) -> Result<Vec<String>>

List all motor topics

Returns all topics under /actuator/motor/*.

pub async fn print_topic_summary(&self) -> Result<()>

Print a formatted summary of all active topics

Useful for debugging and introspection. Groups topics by category and displays them in a human-readable format.

Example

use mecha10::prelude::*;

ctx.print_topic_summary().await?;
// Output:
// === Active Topics ===
// Sensors (3):
//   - /sensor/camera/front
//   - /sensor/lidar/main
//   - /sensor/imu/base
// Actuators (2):
//   - /actuator/motor/left
//   - /actuator/motor/right
// =====================

Trait Implementations

impl AuthExt for Context

async fn register_policy(&self, policy: &Policy) -> Result<()>

Register a policy

async fn get_policy(&self, role: &str) -> Result<Option<Policy>>

Get a policy by role name

async fn list_policies(&self) -> Result<Vec<Policy>>

List all registered policies

async fn assign_role(&self, node_id: &str, role: &str) -> Result<()>

Assign a role to a node

async fn get_role(&self, node_id: &str) -> Result<Option<String>>

Get the role assigned to a node

async fn check_permission( &self, node_id: &str, permission: Permission, topic: &str, ) -> Result<bool>

Check if a node has permission to perform an action on a topic

async fn delete_policy(&self, role: &str) -> Result<()>

Delete a policy

async fn unassign_role(&self, node_id: &str) -> Result<()>

Remove role assignment from a node

impl Clone for Context

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl ContextPublishExt for Context

fn publish_with<'a, 'b, T>( &'a self, topic: Topic<T>, message: &'b T, ) -> PublishBuilder<'a, 'b, T>

where T: Message + Serialize,

Start building a conditional publish operation

impl DiscoveryExt for Context

async fn register(&self, metadata: NodeMetadata) -> Result<()>

Register this node in the service registry

async fn unregister(&self) -> Result<()>

Unregister this node from the service registry

async fn heartbeat(&self) -> Result<()>

Send a heartbeat to update last seen timestamp

async fn discover_all(&self) -> Result<Vec<NodeMetadata>>

Discover all registered nodes

async fn discover_by_type(&self, node_type: &str) -> Result<Vec<NodeMetadata>>

Discover nodes by type

async fn discover_publishers(&self, topic: &str) -> Result<Vec<NodeMetadata>>

Discover who publishes to a specific topic

async fn discover_subscribers(&self, topic: &str) -> Result<Vec<NodeMetadata>>

Discover who subscribes to a specific topic

async fn get_node(&self, node_id: &str) -> Result<Option<NodeMetadata>>

Get metadata for a specific node

impl HealthReportingExt for Context

fn start_health_reporting<'life0, 'async_trait, F, Fut>( &'life0 self, interval: Duration, health_check_fn: F, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where F: Fn() -> Fut + Send + Sync + 'static + 'async_trait, Fut: Future<Output = HealthStatus> + Send + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Start automatic health reporting

impl RpcBreakerExt for Context

fn request_with_breaker<Req, Resp>( &self, topic: &str, request: &Req, config: CircuitBreakerConfig, ) -> BreakerRequestBuilder<'_, Req, Resp>

where Req: Message + Serialize + Clone, Resp: Message + DeserializeOwned + Send + 'static,

Make an RPC request with circuit breaker protection

impl RpcExt for Context

fn request<Req, Resp>( &self, topic: &str, request: &Req, ) -> RequestBuilder<'_, Req, Resp>

where Req: Message + Serialize + Clone, Resp: Message + DeserializeOwned + Send + 'static,

Send a request and wait for response

async fn respond<Req, Resp, F, Fut>( &self, topic: &str, handler: F, ) -> Result<()>

where Req: Message + DeserializeOwned + Send + 'static, Resp: Message + Serialize + Send + 'static, F: Fn(Req) -> Fut + Send + 'static, Fut: Future<Output = Result<Resp>> + Send,

Respond to requests on a topic

async fn subscribe_raw<T>(&self, topic: &str) -> Result<Receiver<T>>

where T: Message + DeserializeOwned + Send + 'static,

Low-level: Subscribe to a topic without type-safe wrapper

async fn publish_raw<T>(&self, topic: &str, message: &T) -> Result<()>

where T: Message + Serialize,

Low-level: Publish to a topic without type-safe wrapper

impl SchemaRegistryExt for Context

async fn register_schema(&self, schema: &MessageSchema) -> Result<()>

Register a schema in the registry

async fn get_schema( &self, name: &str, version: u32, ) -> Result<Option<MessageSchema>>

Get a specific version of a schema

async fn get_latest_schema(&self, name: &str) -> Result<Option<MessageSchema>>

Get the latest version of a schema

async fn list_schemas(&self) -> Result<Vec<MessageSchema>>

List all registered schemas (returns latest version of each)

async fn get_schema_versions(&self, name: &str) -> Result<Vec<MessageSchema>>

Get all versions of a schema

async fn delete_schema(&self, name: &str, version: Option<u32>) -> Result<()>

Delete a schema version

impl SchemaVersioningExt for Context

async fn check_compatibility( &self, name: &str, old_version: u32, new_version: u32, mode: CompatibilityMode, ) -> Result<CompatibilityCheck>

Check compatibility between two schema versions

async fn register_versioned_schema( &self, schema: &MessageSchema, mode: CompatibilityMode, ) -> Result<CompatibilityCheck>

Register a new schema version with compatibility check

async fn get_version_info( &self, name: &str, ) -> Result<Option<MessageVersionInfo>>

Get version info for a message type

async fn deprecate_version( &self, name: &str, version: u32, message: String, ) -> Result<()>

Mark a schema version as deprecated

async fn is_version_deprecated(&self, name: &str, version: u32) -> Result<bool>

Check if using a deprecated version

impl StreamingRpcExt for Context

async fn stream_request<Req, Resp>( &self, topic: &str, request: &Req, ) -> Result<StreamReceiver<Resp>>

where Req: Message + Serialize + Clone, Resp: Message + DeserializeOwned + Send + 'static,

Create a server-streaming RPC (client sends one request, server streams responses)

async fn stream_respond<Req, Resp, F, Fut>( &self, topic: &str, handler: F, ) -> Result<()>

where Req: Message + DeserializeOwned + Send + 'static, Resp: Message + Serialize + Clone + Send + 'static, F: Fn(Req, StreamSender<Resp>) -> Fut + Send + Sync + 'static, Fut: Future<Output = Result<()>> + Send + 'static,

Handle server-streaming RPC requests

async fn bidirectional_stream<T>( &self, topic: &str, ) -> Result<(StreamSender<T>, StreamReceiver<T>)>

where T: Message + Serialize + DeserializeOwned + Clone + Send + 'static,

Create a bidirectional stream (both sides can send/receive)

impl TracingExt for Context

async fn publish_traced<T: Message + Serialize + Clone + 'static>( &self, topic: &str, message: &T, ) -> Result<()>

Publish a message with automatic trace context

async fn publish_with_context<T: Message + Serialize + Clone + 'static>( &self, topic: &str, message: &T, trace_context: TraceContext, ) -> Result<()>

Publish with explicit trace context (for continuing traces)

async fn subscribe_traced<T: Message + DeserializeOwned + Send + 'static>( &self, topic: &str, ) -> Result<Receiver<TracedMessage<T>>>

Subscribe with trace context propagation

fn current_trace_context(&self) -> Option<TraceContext>

Get current trace context from span (if any)

fn extract_trace_context( &self, metadata: &HashMap<String, String>, ) -> Option<TraceContext>

Extract trace context from message metadata

fn inject_trace_context( &self, metadata: &mut HashMap<String, String>, ctx: &TraceContext, )

Inject trace context into message metadata