Struct Context 

pub struct Context { /* private fields */ }

Implementations

impl Context

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

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_core::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>, Mecha10Error>

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_core::prelude::*;
use mecha10_core::messages::Image;

// Subscribe to all cameras with one call
let mut cameras = ctx.subscribe_aggregated::<Image>("/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 subscribe_pattern<T>( &self, topic: Topic<T>, pattern: &str, ) -> Result<AggregatedReceiver<T>, Mecha10Error>

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_core::prelude::*;
use mecha10_core::topics::sensor;

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

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 fn get_redis_url() -> Result<String, Mecha10Error>

Get Redis URL from infrastructure configuration

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

Get control plane Redis URL from infrastructure configuration

impl Context

pub async fn new(node_id: &str) -> Result<Context, Mecha10Error>

Create a new context

Arguments

• node_id - Unique identifier for this node (fallback if NODE_NAME env var not set)

Environment Variables

• NODE_NAME - If set, overrides the node_id parameter. This allows node-runner to set full identifiers like @mecha10/speaker for consistent naming across the CLI, dashboard, and health reporting.

Example

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

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

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 node_id(&self) -> &str

Get the node ID

pub async fn has_control_plane(&self) -> bool

Check if control plane Redis connection is available

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_core::prelude::*;

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

impl Context

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

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), selected within a config file’s {"dev": {...}, "production": {...}} envelope (see Self::load_environment_aware_config)
• File-level fallback (no key-by-key merging)
• Automatic project-root discovery via crate::fs_utils::find_project_root

Tier order (mirrors mecha10_cli_core::paths::config::node_config_candidates, the canonical 2-tier resolution order for node configs – this method can’t call that helper directly since mecha10-core is a dependency of mecha10-cli-core, not the reverse, but it must stay in sync with it):

1. configs/nodes/@mecha10/{node_name}/config.json - Framework nodes
2. configs/nodes/{node_name}/config.json - Project nodes (preferred)
3. T::default() - Default implementation, if neither tier has a config file (or no project root could be found)

Environment Detection:

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

Semantic Validation:

• Not run here. This method is generic over T with only a DeserializeOwned + Default bound, and stable Rust has no specialization – a function generic over T can’t conditionally call crate::config::Validate::validate only when T happens to implement it. Node types that need semantic validation should call validate() explicitly on the config returned here.

Type Parameters

• T - Configuration type that implements DeserializeOwned + Default

Arguments

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

Returns

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

Example

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

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

let config: SimBridgeConfig = ctx.load_node_config("simulator").await?;

pub async fn load_node_config_for_instance<T>( &self, node_name: &str, instance_id: Option<&str>, ) -> Result<T, Mecha10Error>

where T: DeserializeOwned + Default,

Like Self::load_node_config, but with an explicit instance_id instead of this context’s own Self::instance.

Node-runner (the supervisor process) is the main reason this is split out: it loads a node’s config.json on that node’s behalf, through a context that represents node-runner itself (not the node), so it can’t rely on self.instance() - it passes the target node’s --instance-id through explicitly instead. Every actual node binary (via node_main!/#[derive(Node)]) should just call Self::load_node_config, which already does the right thing from its own instance().

Instance-Keyed Config Files

For a multi-instance node (two camera nodes - one real, one simulated - both declared in mecha10.json via the {"name": "@mecha10/camera", "instance_id": "real"} node-list entry shape, LAB-4131), the resolved config file’s per-environment section may itself be a flat config object (today’s behavior, unchanged, used when instance_id is None) or a JSON object keyed by instance id:

{
  "dev": {
    "real": { "source": { "type": "v4l2", "device_path": "/dev/video0" }, "id": "real" },
    "sim":  { "source": { "type": "redis", "topic": "/camera/sim/raw" }, "id": "sim" }
  }
}

When instance_id is Some(id), the sub-object at that key is used as the node’s config (same shape/validation as today’s flat config). An instance_id that isn’t present as a key is an error, rather than silently falling back to T::default() - a multi-instance node whose config is missing an instance’s entry has a real misconfiguration, not an absent config file.

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

where T: DeserializeOwned + Default + JsonSchema,

Same as Self::load_node_config, but also validates the loaded JSON against T’s own schemars-derived JSON Schema before deserializing into it (LAB-4066).

This is the load-at-startup counterpart to mecha10_core::schema_validation’s preflight validation (validate_project_config/mecha10 config validate), using the same “every node supplies its own schema” principle that validation’s declared-schema lookup is built on – except here there is no lookup and no “missing schema” case to hard-error on: T: JsonSchema means T’s own derived schema always exists, enforced by the type system at compile time rather than by a runtime registry/convention check. node_main! and #[derive(Node)]’s generated run() call this (not Self::load_node_config) so every node – built-in or custom – gets this check for free the moment its Config struct derives JsonSchema (already required to export a schema for it at all, see mecha10-schema-registry).

Not used for Self::load_node_config callers that read another node’s config file generically through a deliberately-partial/lenient struct (e.g. mecha10-node-runner’s launcher-level NodeConfig) – validating a partial struct’s derived schema against another node’s full config would reject fields the reader never claimed to model, which isn’t this node’s schema to enforce.

Errors

Returns an error if the config file exists but fails to parse, fails schema validation, or fails to deserialize into T.

impl Context

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

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_core::prelude::*;
use mecha10_core::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<(), Mecha10Error>

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_core::prelude::*;
use mecha10_core::topics::sensor;
use mecha10_core::messages::Image;

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

// 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<(), Mecha10Error>

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<(), Mecha10Error>

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_core::prelude::*;
use mecha10_core::topics::sensor;
use mecha10_core::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![],
    frame_id: None,
};

// 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<(), Mecha10Error>

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_core::prelude::*;
use mecha10_core::topics::actuator;

// Velocity + torque PWM values for this motor
let cmd = vec![1.0, 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>, Mecha10Error>

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_core::prelude::*;
use mecha10_core::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>, Mecha10Error>

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 publish_raw<T>( &self, topic_path: &str, message: &T, ) -> Result<(), Mecha10Error>

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.

impl Context

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

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_core::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_core::prelude::*;
use mecha10_core::state::ConcreteStateManager;
use std::sync::Arc;

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

impl Context

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

Subscribe to all camera topics

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

Example

use mecha10_core::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>, Mecha10Error>

Subscribe to all sensor topics

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

Example

use mecha10_core::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>, Mecha10Error>

Subscribe to all LiDAR topics

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

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

Subscribe to all IMU topics

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

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

Subscribe to all motor topics

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

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

List all sensor topics currently active in the system

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

Example

use mecha10_core::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>, Mecha10Error>

List all actuator topics currently active in the system

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

Example

use mecha10_core::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>, Mecha10Error>

List all topics in the system

Returns all active topics regardless of category.

Example

use mecha10_core::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>, Mecha10Error>

List all camera topics

Returns all topics under /sensor/camera/*.

Example

use mecha10_core::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>, Mecha10Error>

List all LiDAR topics

Returns all topics under /sensor/lidar/*.

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

List all IMU topics

Returns all topics under /sensor/imu/*.

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

List all motor topics

Returns all topics under /actuator/motor/*.

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

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_core::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<(), Mecha10Error>

Register a policy

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

Get a policy by role name

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

List all registered policies

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

Assign a role to a node

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

Get the role assigned to a node

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

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

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

Delete a policy

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

Remove role assignment from a node

impl Clone for Context

fn clone(&self) -> Context

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<(), Mecha10Error>

Register this node in the service registry

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

Unregister this node from the service registry

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

Send a heartbeat to update last seen timestamp

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

Discover all registered nodes

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

Discover nodes by type

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

Discover who publishes to a specific topic

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

Discover who subscribes to a specific topic

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

Get metadata for a specific node

impl HealthReportingExt for Context

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

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

Start automatic health reporting with default settings (Normal priority, 5s interval)

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

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

Start automatic health reporting with custom configuration

fn start_health_reporting_with_type_tags<'life0, 'async_trait, F, Fut>( &'life0 self, node_type: Option<NodeType>, tags: &'static [NodeTag], health_check_fn: F, ) -> Pin<Box<dyn Future<Output = Result<(), Mecha10Error>> + Send + 'async_trait>>

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

Start automatic health reporting with default settings (Normal priority, 5s interval), tagged with an explicit node_type/tags pair.

fn start_health_reporting_with_config_and_type_tags<'life0, 'async_trait, F, Fut>( &'life0 self, priority: HealthPriority, interval: Duration, node_type: Option<NodeType>, tags: &'static [NodeTag], health_check_fn: F, ) -> Pin<Box<dyn Future<Output = Result<(), Mecha10Error>> + Send + 'async_trait>>

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

Start automatic health reporting with custom configuration, tagged with an explicit node_type/tags pair. See HealthReportingExt::start_health_reporting_with_type_tags.

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<(), Mecha10Error>

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

Respond to requests on a topic

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

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<(), Mecha10Error>

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<(), Mecha10Error>

Register a schema in the registry

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

Get a specific version of a schema

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

Get the latest version of a schema

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

List all registered schemas (returns latest version of each)

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

Get all versions of a schema

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

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, Mecha10Error>

Check compatibility between two schema versions

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

Register a new schema version with compatibility check

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

Get version info for a message type

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

Mark a schema version as deprecated

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

Check if using a deprecated version

impl StreamingRpcExt for Context

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

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<(), Mecha10Error>

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<(), Mecha10Error>> + Send + 'static,

Handle server-streaming RPC requests

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

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>( &self, topic: &str, message: &T, ) -> Result<(), Mecha10Error>

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

Publish a message with automatic trace context

async fn publish_with_context<T>( &self, topic: &str, message: &T, trace_context: TraceContext, ) -> Result<(), Mecha10Error>

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

Publish with explicit trace context (for continuing traces)

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

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

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