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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
//! `map`, `filter`, and `filter_map` combinators.
use super::relay::RelayAction;
use super::types::{Receiver, ReceiverInner};
use tokio::sync::mpsc;
enum MappedSender<T> {
Unbounded(mpsc::UnboundedSender<T>),
Bounded(mpsc::Sender<T>),
}
impl<T> Receiver<T> {
/// 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
///
/// ```rust
/// # use mecha10_core::prelude::*;
/// # use mecha10_core::topics::{actuator, sensor};
/// # async fn example(ctx: &Context) -> Result<()> {
/// // 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);
/// }
/// # Ok(())
/// # }
/// # fn resize_image(img: mecha10_core::messages::Image, w: u32, h: u32) -> mecha10_core::messages::Image { img }
/// ```
pub fn map<U, F>(self, f: F) -> Receiver<U>
where
F: FnMut(T) -> U + Send + 'static,
T: Send + 'static,
U: Send + 'static,
{
let (tx, rx) = match &self.inner {
ReceiverInner::Unbounded(_) => {
let (tx, rx) = mpsc::unbounded_channel();
(MappedSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
}
ReceiverInner::Bounded(bounded_rx) => {
let capacity = bounded_rx.max_capacity();
let (tx, rx) = mpsc::channel(capacity);
(MappedSender::Bounded(tx), ReceiverInner::Bounded(rx))
}
ReceiverInner::DropOldest(drop_oldest_rx) => {
let capacity = drop_oldest_rx.capacity();
let (tx, rx) = mpsc::channel(capacity);
(MappedSender::Bounded(tx), ReceiverInner::Bounded(rx))
}
};
let mut receiver = self;
let mut transform = f;
tokio::spawn(async move {
while let Some(msg) = receiver.recv().await {
let transformed = transform(msg);
let send_result = match tx {
MappedSender::Unbounded(ref tx) => tx.send(transformed).map_err(|_| ()),
MappedSender::Bounded(ref tx) => tx.send(transformed).await.map_err(|_| ()),
};
if send_result.is_err() {
break;
}
}
});
Receiver { inner: rx }
}
/// 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
///
/// ```rust
/// # use mecha10_core::prelude::*;
/// # use mecha10_core::topics::{actuator, sensor};
/// # async fn example(ctx: &Context) -> Result<()> {
/// // 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
/// }
/// # Ok(())
/// # }
/// ```
pub fn filter<F>(self, mut predicate: F) -> Receiver<T>
where
F: FnMut(&T) -> bool + Send + 'static,
T: Send + 'static,
{
self.spawn_relay(move |msg| {
if predicate(&msg) {
RelayAction::Forward(msg)
} else {
RelayAction::Drop
}
})
}
/// 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
///
/// ```rust
/// # use mecha10_core::prelude::*;
/// # use mecha10_core::topics::{actuator, system};
/// # async fn example(ctx: &Context) -> Result<()> {
/// // 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
/// }
/// # Ok(())
/// # }
/// # fn parse_sensor_data(raw: &String) -> Result<String> { Ok(raw.clone()) }
/// ```
pub fn filter_map<U, F>(self, mut f: F) -> Receiver<U>
where
F: FnMut(T) -> Option<U> + Send + 'static,
T: Send + 'static,
U: Send + 'static,
{
self.spawn_relay(move |msg| match f(msg) {
Some(transformed) => RelayAction::Forward(transformed),
None => RelayAction::Drop,
})
}
}