mecha10_core/context/receiver/time_combinators.rs
1//! `debounce` and `window` time-based combinators.
2//!
3//! duplicate: false positive — cargo-dupes groups `debounce`/`window` (here) with
4//! `Receiver::spawn_relay` (relay.rs) as an exact duplicate. The part that's genuinely shared
5//! across all three — creating an output channel matching `self`'s boundedness, spawning a
6//! background task, and wrapping the result back into a `Receiver` — is already factored out
7//! into `matching_channel` and used by all of them (see relay.rs). What's left in `debounce`/
8//! `window` is a `tokio::select!` loop reacting to both incoming messages *and* a timer, which
9//! is structurally unlike `spawn_relay`'s per-item-only loop and unlike each other in its
10//! actual state machine (quiet-period debounce vs. tumbling-window batching). Forcing those two
11//! into one generic time-combinator would add a real risk of subtly changing the concurrency
12//! behavior of both for a purely mechanical dedup — not a change to make outside a dedicated
13//! review of this module.
14
15use super::relay::matching_channel;
16use super::types::Receiver;
17use std::time::Duration;
18use tokio::time::{sleep_until, Instant};
19
20impl<T> Receiver<T> {
21 /// Debounce (rate-limit) messages to prevent rapid bursts
22 ///
23 /// Creates a new receiver that only emits a message if a certain duration has passed
24 /// since the last emission. This is useful for reducing update frequency of high-rate
25 /// sensors or preventing UI update storms.
26 ///
27 /// **Strategy:** "Emit latest after quiet period"
28 /// - Messages are buffered during the debounce period
29 /// - Only the most recent message is emitted after the period expires
30 /// - Earlier messages within the period are dropped
31 ///
32 /// # Arguments
33 ///
34 /// * `duration` - Minimum time between emissions
35 ///
36 /// # Example
37 ///
38 /// ```rust
39 /// # use mecha10_core::prelude::*;
40 /// # use mecha10_core::topics::{actuator, sensor};
41 /// # use std::time::Duration;
42 /// # async fn example(ctx: &Context) -> Result<()> {
43 /// // Reduce camera updates to max 10 Hz (100ms debounce)
44 /// let mut debounced_images = ctx.subscribe(sensor::CAMERA_RGB)
45 /// .await?
46 /// .debounce(Duration::from_millis(100));
47 ///
48 /// while let Some(image) = debounced_images.recv().await {
49 /// // Process at most 10 images/sec
50 /// }
51 ///
52 /// // Prevent rapid velocity command updates
53 /// let mut stable_status = ctx.subscribe(actuator::CMD_VEL)
54 /// .await?
55 /// .debounce(Duration::from_millis(50));
56 ///
57 /// while let Some(status) = stable_status.recv().await {
58 /// // Only get updates every 50ms
59 /// }
60 /// # Ok(())
61 /// # }
62 /// ```
63 pub fn debounce(self, duration: Duration) -> Receiver<T>
64 where
65 T: Send + 'static,
66 {
67 let (tx, rx) = matching_channel::<T, T>(&self.inner);
68
69 let mut receiver = self;
70 tokio::spawn(async move {
71 let mut deadline: Option<Instant> = None;
72 let mut pending_msg: Option<T> = None;
73 loop {
74 tokio::select! {
75 msg = receiver.recv() => {
76 match msg {
77 Some(new_msg) => {
78 pending_msg = Some(new_msg);
79 deadline = Some(Instant::now() + duration);
80 }
81 None => {
82 if let Some(msg_to_send) = pending_msg.take() {
83 let _ = tx.send(msg_to_send).await;
84 }
85 break;
86 }
87 }
88 }
89
90 _ = sleep_until(deadline.unwrap_or_else(Instant::now)), if deadline.is_some() => {
91 if let Some(msg_to_send) = pending_msg.take() {
92 if tx.send(msg_to_send).await.is_err() {
93 break;
94 }
95 }
96
97 deadline = None;
98 }
99 }
100 }
101 });
102
103 Receiver { inner: rx }
104 }
105
106 /// Window messages over a time period
107 ///
108 /// Creates a new receiver that collects messages over a fixed time window and emits
109 /// them as a batch (Vec) at the end of each window. This is useful for:
110 /// - Aggregating sensor readings (averaging, min/max)
111 /// - Batch processing for efficiency
112 /// - Time-series analysis
113 /// - Rate calculation (count messages per window)
114 ///
115 /// # Behavior
116 ///
117 /// - **Fixed Windows**: Each window has a fixed duration
118 /// - **Non-overlapping**: Windows don't overlap (tumbling window)
119 /// - **Batch Emission**: All messages in window emitted as Vec at window end
120 /// - **Empty Windows**: Empty Vec emitted if no messages received
121 /// - **Termination**: Emits final partial window when source ends
122 ///
123 /// # Arguments
124 ///
125 /// * `window_duration` - Duration of each time window
126 ///
127 /// # Example
128 ///
129 /// ```rust
130 /// # use mecha10_core::prelude::*;
131 /// # use mecha10_core::messages::Image;
132 /// # use mecha10_core::topics::sensor;
133 /// # use std::time::Duration;
134 /// # async fn example(ctx: &Context) -> Result<()> {
135 /// // Average IMU readings over 100ms windows
136 /// let mut windowed = ctx.subscribe(sensor::IMU)
137 /// .await?
138 /// .window(Duration::from_millis(100));
139 ///
140 /// while let Some(batch) = windowed.recv().await {
141 /// if !batch.is_empty() {
142 /// let avg = batch.iter().map(|imu| imu.linear_acceleration[0]).sum::<f32>() / batch.len() as f32;
143 /// println!("Average accel: {} (n={})", avg, batch.len());
144 /// }
145 /// }
146 ///
147 /// // Count messages per second
148 /// let mut counts = ctx.subscribe(sensor::BATTERY_VOLTAGE)
149 /// .await?
150 /// .window(Duration::from_secs(1))
151 /// .map(|batch| batch.len());
152 ///
153 /// while let Some(count) = counts.recv().await {
154 /// println!("Events/sec: {}", count);
155 /// }
156 ///
157 /// // Batch process for efficiency
158 /// let mut batches = ctx.subscribe(sensor::CAMERA_RGB)
159 /// .await?
160 /// .window(Duration::from_millis(50));
161 ///
162 /// while let Some(batch) = batches.recv().await {
163 /// // Process entire batch at once (more efficient than one-by-one)
164 /// process_batch(&batch).await?;
165 /// }
166 /// # Ok(())
167 /// # }
168 /// # async fn process_batch(batch: &[Image]) -> Result<()> { Ok(()) }
169 /// ```
170 ///
171 /// # Performance
172 ///
173 /// - **Memory**: Buffers all messages within window (use reasonable window sizes)
174 /// - **Latency**: Adds latency equal to window duration
175 /// - **CPU**: Minimal overhead for window management
176 ///
177 /// # Caveats
178 ///
179 /// - Window size affects memory usage (smaller windows = less memory)
180 /// - All messages are delivered at window end (batch latency)
181 /// - For sliding windows, consider multiple overlapping `.window()` subscriptions
182 pub fn window(self, window_duration: Duration) -> Receiver<Vec<T>>
183 where
184 T: Send + 'static,
185 {
186 let (tx, rx) = matching_channel::<T, Vec<T>>(&self.inner);
187
188 let mut receiver = self;
189 tokio::spawn(async move {
190 let mut buffer: Vec<T> = Vec::new();
191 let mut interval = tokio::time::interval(window_duration);
192 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
193
194 loop {
195 tokio::select! {
196 msg = receiver.recv() => {
197 match msg {
198 Some(new_msg) => {
199 buffer.push(new_msg);
200 }
201 None => {
202 if !buffer.is_empty() {
203 let _ = tx.send(buffer).await;
204 }
205 break;
206 }
207 }
208 }
209
210 _ = interval.tick() => {
211 let batch = std::mem::take(&mut buffer);
212 if tx.send(batch).await.is_err() {
213 break;
214 }
215 }
216 }
217 }
218 });
219
220 Receiver { inner: rx }
221 }
222}