mecha10_core/context/receiver/
zip.rs

1//! `zip` combinator: pairing two receivers into one combined stream.
2
3use super::types::{Receiver, ReceiverInner};
4use tokio::sync::mpsc;
5
6enum ZippedSender<T> {
7    Unbounded(mpsc::UnboundedSender<T>),
8    Bounded(mpsc::Sender<T>),
9}
10
11impl<T> Receiver<T> {
12    /// Combine two streams by pairing their messages
13    ///
14    /// Creates a new stream that emits pairs of messages from two input streams.
15    /// The combined stream emits `(T, U)` tuples when both streams have messages available.
16    ///
17    /// # Behavior
18    ///
19    /// - **Paired Emission**: Waits for both streams to have messages, then emits the pair
20    /// - **Buffering**: Buffers messages from faster stream until slower stream catches up
21    /// - **Termination**: Stops when either stream ends
22    /// - **Capacity**: Uses the smaller capacity of the two input streams
23    ///
24    /// # Use Cases
25    ///
26    /// - Combining sensor readings (e.g., camera + LiDAR)
27    /// - Synchronizing multiple data sources
28    /// - Correlating events from different streams
29    /// - Sensor fusion
30    ///
31    /// # Example
32    ///
33    /// ```rust
34    /// use mecha10_core::prelude::*;
35    /// use mecha10_core::messages::{Image, LaserScan};
36    /// use mecha10_core::topics::sensor;
37    ///
38    /// # async fn example(ctx: &Context) -> Result<()> {
39    /// // Subscribe to two sensor streams
40    /// let camera = ctx.subscribe(sensor::CAMERA_RGB).await?;
41    /// let lidar = ctx.subscribe(sensor::LIDAR_SCAN).await?;
42    ///
43    /// // Combine them
44    /// let mut fused = camera.zip(lidar);
45    ///
46    /// // Process paired messages
47    /// while let Some((img, scan)) = fused.recv().await {
48    ///     // Both sensors have data - process together
49    ///     process_sensor_fusion(&img, &scan).await?;
50    /// }
51    /// # Ok(())
52    /// # }
53    /// # async fn process_sensor_fusion(img: &Image, scan: &LaserScan) -> Result<()> { Ok(()) }
54    /// ```
55    ///
56    /// # Performance
57    ///
58    /// - **Memory**: Buffers messages from faster stream (bounded by capacity)
59    /// - **Latency**: Adds latency equal to slower stream's inter-message delay
60    /// - **CPU**: Minimal overhead for pairing logic
61    ///
62    /// # Caveats
63    ///
64    /// - If streams have very different rates, slower stream will be the bottleneck
65    /// - Buffer can fill up if rate difference is sustained
66    /// - Consider `.debounce()` on faster stream if rate mismatch is severe
67    pub fn zip<U>(self, other: Receiver<U>) -> Receiver<(T, U)>
68    where
69        T: Send + 'static,
70        U: Send + 'static,
71    {
72        fn effective_capacity<X>(inner: &ReceiverInner<X>) -> Option<usize> {
73            match inner {
74                ReceiverInner::Unbounded(_) => None,
75                ReceiverInner::Bounded(rx) => Some(rx.max_capacity()),
76                ReceiverInner::DropOldest(rx) => Some(rx.capacity()),
77            }
78        }
79
80        let capacity = match (effective_capacity(&self.inner), effective_capacity(&other.inner)) {
81            (None, None) => None,
82            (Some(cap), None) | (None, Some(cap)) => Some(cap),
83            (Some(a), Some(b)) => Some(a.min(b)),
84        };
85
86        let (tx, rx) = if let Some(cap) = capacity {
87            let (tx, rx) = mpsc::channel(cap);
88            (ZippedSender::Bounded(tx), ReceiverInner::Bounded(rx))
89        } else {
90            let (tx, rx) = mpsc::unbounded_channel();
91            (ZippedSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
92        };
93
94        let mut receiver_a = self;
95        let mut receiver_b = other;
96
97        tokio::spawn(async move {
98            loop {
99                let msg_a = receiver_a.recv().await;
100                let msg_b = receiver_b.recv().await;
101
102                match (msg_a, msg_b) {
103                    (Some(a), Some(b)) => {
104                        let send_result = match &tx {
105                            ZippedSender::Unbounded(tx) => tx.send((a, b)).map_err(|_| ()),
106                            ZippedSender::Bounded(tx) => tx.send((a, b)).await.map_err(|_| ()),
107                        };
108
109                        if send_result.is_err() {
110                            break;
111                        }
112                    }
113                    _ => {
114                        break;
115                    }
116                }
117            }
118        });
119
120        Receiver { inner: rx }
121    }
122}
mecha10_core/context/receiver/zip.rs

mecha10_core/context/receiver/
zip.rs
