1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
//! Pattern-based topic discovery and aggregated (multi-topic) subscriptions.
#![cfg(feature = "messaging")]
use super::Context;
use crate::error::{Mecha10Error, Result};
use crate::messages::Message;
use anyhow::Context as _;
use serde::de::DeserializeOwned;
impl Context {
/// 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
///
/// ```rust,no_run
/// use mecha10_core::prelude::*;
///
/// # async fn example(ctx: &Context) -> Result<()> {
/// // Discover all camera topics
/// let topics = ctx.discover_topics("/sensor/camera/*").await?;
/// println!("Found {} camera topics: {:?}", topics.len(), topics);
/// # Ok(())
/// # }
/// ```
pub async fn discover_topics(&self, pattern: &str) -> Result<Vec<String>> {
use crate::aggregated::pattern::convert_glob_to_redis;
let bus = self.message_bus.read().await;
let redis_pattern = convert_glob_to_redis(pattern);
bus.discover_topics(&redis_pattern)
.await
.with_context(|| {
format!(
"Failed to discover topics matching pattern '{}' (redis pattern: '{}') from node '{}'",
pattern, redis_pattern, self.node_id
)
})
.map_err(|e| Mecha10Error::MessagingError {
message: format!("{:#}", e),
suggestion: "Check that Redis is running and the pattern is valid".to_string(),
})
}
/// 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
///
/// ```rust,no_run
/// use mecha10_core::prelude::*;
/// use mecha10_core::messages::Image;
///
/// # async fn example(ctx: &Context) -> Result<()> {
/// // 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);
/// }
/// # Ok(())
/// # }
/// ```
pub async fn subscribe_aggregated<T>(&self, pattern: &str) -> Result<crate::aggregated::AggregatedReceiver<T>>
where
T: Message + DeserializeOwned + Send + 'static,
{
let matching_topics = self.discover_topics(pattern).await?;
if matching_topics.is_empty() {
return Err(Mecha10Error::MessagingError {
message: format!("No topics found matching pattern: {}", pattern),
suggestion: "Check that the pattern is correct and topics exist".to_string(),
});
}
let mut receivers = Vec::new();
for topic_path in matching_topics {
let rx = self.subscribe_raw::<T>(&topic_path).await?;
let receiver_type = match rx.inner {
crate::context::receiver::ReceiverInner::Unbounded(unbounded) => {
crate::aggregated::ReceiverType::Unbounded(unbounded)
}
crate::context::receiver::ReceiverInner::Bounded(bounded) => {
crate::aggregated::ReceiverType::Bounded(bounded)
}
crate::context::receiver::ReceiverInner::DropOldest(mut drop_oldest) => {
let capacity = drop_oldest.capacity();
let (tx, bounded) = tokio::sync::mpsc::channel(capacity.max(1));
tokio::spawn(async move {
while let Some(msg) = drop_oldest.recv().await {
if tx.send(msg).await.is_err() {
break;
}
}
});
crate::aggregated::ReceiverType::Bounded(bounded)
}
};
receivers.push((topic_path.clone(), receiver_type));
}
Ok(crate::aggregated::AggregatedReceiver::new_mixed(
pattern.to_string(),
receivers,
))
}
/// Subscribe to a topic pattern with wildcard
///
/// Generic helper for subscribing to multiple topics matching a pattern.
///
/// # Example
///
/// ```rust,no_run
/// use mecha10_core::prelude::*;
/// use mecha10_core::topics::sensor;
///
/// # async fn example(ctx: &Context) -> Result<()> {
/// // Subscribe to all instances of a specific topic type
/// let mut cameras = ctx.subscribe_pattern(sensor::CAMERA_RGB, "*").await?;
/// # Ok(())
/// # }
/// ```
pub async fn subscribe_pattern<T>(
&self,
topic: crate::topics::Topic<T>,
pattern: &str,
) -> Result<crate::aggregated::AggregatedReceiver<T>>
where
T: Message + DeserializeOwned + Send + 'static,
{
let full_pattern = format!("{}/{}", topic.path(), pattern);
self.subscribe_aggregated(&full_pattern).await
}
}