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
//! `StreamBuilder::window`: time-windowed batching.
use super::StreamBuilder;
use crate::context::Receiver;
use crate::messages::Message;
use std::time::Duration;
use tokio::sync::mpsc;
impl<T: Message + Send + 'static> StreamBuilder<T> {
/// Window messages over a time period
///
/// Collects messages over a fixed time window and emits them as a vector.
/// This is useful for time-based aggregation and analysis.
///
/// # Arguments
///
/// * `window_duration` - Duration of the time window
///
/// # Returns
///
/// A `Receiver<Vec<T>>` that yields vectors of messages collected in each window
///
/// # Example
///
/// ```rust
/// # use mecha10_core::prelude::*;
/// # async fn example(ctx: &Context) -> Result<()> {
/// // Collect detections over 1-second windows
/// let mut windowed = ctx.subscribe(perception::DETECTIONS)
/// .window(Duration::from_secs(1));
///
/// while let Some(detections) = windowed.recv().await {
/// // Process all detections from the last second
/// println!("Received {} detections in 1 second window", detections.len());
///
/// // Example: Calculate average confidence
/// let avg_confidence = detections.iter()
/// .map(|d| d.confidence)
/// .sum::<f32>() / detections.len() as f32;
/// println!("Average confidence: {:.2}", avg_confidence);
/// }
/// # Ok(())
/// # }
/// ```
///
/// # Use Cases
///
/// - **Rate calculation**: Count messages per time window
/// - **Moving averages**: Compute statistics over windows
/// - **Burst detection**: Identify spikes in message frequency
/// - **Time-series analysis**: Analyze temporal patterns
pub fn window(self, window_duration: Duration) -> Receiver<Vec<T>>
where
T: Clone,
{
let (tx, rx) = mpsc::unbounded_channel();
tokio::spawn(async move {
Self::run_window_pipeline(self.receiver, tx, window_duration).await;
});
Receiver::from(rx)
}
/// Internal method to run the window pipeline
async fn run_window_pipeline(
mut receiver: Receiver<T>,
tx: mpsc::UnboundedSender<Vec<T>>,
window_duration: Duration,
) where
T: Clone,
{
let mut window = Vec::new();
let mut deadline = tokio::time::Instant::now() + window_duration;
loop {
tokio::select! {
// Receive next message
msg_opt = receiver.recv() => {
match msg_opt {
Some(msg) => {
window.push(msg);
}
None => {
// Stream ended - send remaining window if any
if !window.is_empty() {
let _ = tx.send(window);
}
break;
}
}
}
// Window timeout - emit collected messages
_ = tokio::time::sleep_until(deadline) => {
// Send window (even if empty for consistent timing)
if tx.send(window.clone()).is_err() {
break; // Receiver dropped
}
window.clear();
deadline = tokio::time::Instant::now() + window_duration;
}
}
}
}
}