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
//! `zip` combinator: pairing two receivers into one combined stream.
use super::types::{Receiver, ReceiverInner};
use tokio::sync::mpsc;
/// Sender type used by [`Receiver::zip`].
enum ZippedSender<T> {
Unbounded(mpsc::UnboundedSender<T>),
Bounded(mpsc::Sender<T>),
}
impl<T> Receiver<T> {
/// Combine two streams by pairing their messages
///
/// Creates a new stream that emits pairs of messages from two input streams.
/// The combined stream emits `(T, U)` tuples when both streams have messages available.
///
/// # Behavior
///
/// - **Paired Emission**: Waits for both streams to have messages, then emits the pair
/// - **Buffering**: Buffers messages from faster stream until slower stream catches up
/// - **Termination**: Stops when either stream ends
/// - **Capacity**: Uses the smaller capacity of the two input streams
///
/// # Use Cases
///
/// - Combining sensor readings (e.g., camera + LiDAR)
/// - Synchronizing multiple data sources
/// - Correlating events from different streams
/// - Sensor fusion
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// # async fn example(ctx: &Context) -> Result<()> {
/// // Subscribe to two sensor streams
/// let camera = ctx.subscribe("/sensor/camera").await?;
/// let lidar = ctx.subscribe("/sensor/lidar").await?;
///
/// // Combine them
/// let mut fused = camera.zip(lidar);
///
/// // Process paired messages
/// while let Some((img, scan)) = fused.recv().await {
/// // Both sensors have data - process together
/// process_sensor_fusion(&img, &scan).await?;
/// }
/// # Ok(())
/// # }
/// ```
///
/// # Performance
///
/// - **Memory**: Buffers messages from faster stream (bounded by capacity)
/// - **Latency**: Adds latency equal to slower stream's inter-message delay
/// - **CPU**: Minimal overhead for pairing logic
///
/// # Caveats
///
/// - If streams have very different rates, slower stream will be the bottleneck
/// - Buffer can fill up if rate difference is sustained
/// - Consider `.debounce()` on faster stream if rate mismatch is severe
pub fn zip<U>(self, other: Receiver<U>) -> Receiver<(T, U)>
where
T: Send + 'static,
U: Send + 'static,
{
// Determine capacity based on both receivers
fn effective_capacity<X>(inner: &ReceiverInner<X>) -> Option<usize> {
match inner {
ReceiverInner::Unbounded(_) => None,
ReceiverInner::Bounded(rx) => Some(rx.max_capacity()),
ReceiverInner::DropOldest(rx) => Some(rx.capacity()),
}
}
let capacity = match (effective_capacity(&self.inner), effective_capacity(&other.inner)) {
(None, None) => None,
(Some(cap), None) | (None, Some(cap)) => Some(cap),
(Some(a), Some(b)) => Some(a.min(b)),
};
// Create output channel
let (tx, rx) = if let Some(cap) = capacity {
let (tx, rx) = mpsc::channel(cap);
(ZippedSender::Bounded(tx), ReceiverInner::Bounded(rx))
} else {
let (tx, rx) = mpsc::unbounded_channel();
(ZippedSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
};
// Spawn task to zip messages
let mut receiver_a = self;
let mut receiver_b = other;
tokio::spawn(async move {
loop {
// Wait for both streams to have messages
let msg_a = receiver_a.recv().await;
let msg_b = receiver_b.recv().await;
match (msg_a, msg_b) {
(Some(a), Some(b)) => {
// Both messages available, send pair
let send_result = match &tx {
ZippedSender::Unbounded(tx) => tx.send((a, b)).map_err(|_| ()),
ZippedSender::Bounded(tx) => tx.send((a, b)).await.map_err(|_| ()),
};
if send_result.is_err() {
// Receiver dropped, stop zipping
break;
}
}
_ => {
// One or both streams ended
break;
}
}
}
});
Receiver { inner: rx }
}
}