mecha10-core 0.8.0

//! `StreamBuilder::zip`: pairing this stream with another.

use super::StreamBuilder;
use crate::context::Receiver;
use crate::messages::Message;
use tokio::sync::mpsc;

impl<T: Message + Send + 'static> StreamBuilder<T> {
    /// Zip this stream with another stream
    ///
    /// Combines two streams element-by-element, producing tuples of messages.
    /// The resulting stream will emit `(T, U)` tuples when both streams have
    /// messages available.
    ///
    /// If one stream ends before the other, the zip stream will also end.
    ///
    /// # Arguments
    ///
    /// * `other` - Another receiver to zip with this stream
    ///
    /// # Returns
    ///
    /// A `Receiver<(T, U)>` that yields tuples of messages from both streams
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10_core::prelude::*;
    /// # use mecha10_core::topics::sensor;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Zip camera images with LiDAR scans
    /// let images = ctx.subscribe(sensor::CAMERA_RGB).await?;
    /// let scans = ctx.subscribe(sensor::LIDAR_SCAN).await?;
    ///
    /// let mut zipped = images.zip(scans);
    ///
    /// while let Some((image, scan)) = zipped.recv().await {
    ///     // Process synchronized image + LiDAR data
    ///     println!("Image {}x{}, LiDAR {} points",
    ///         image.width, image.height, scan.ranges.len());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn zip<U>(self, other: Receiver<U>) -> Receiver<(T, U)>
    where
        U: Message + Send + 'static,
    {
        let (tx, rx) = mpsc::unbounded_channel();

        tokio::spawn(async move {
            Self::run_zip_pipeline(self.receiver, other, tx).await;
        });

        Receiver::from(rx)
    }

    /// Internal method to run the zip pipeline
    async fn run_zip_pipeline<U>(
        mut receiver1: Receiver<T>,
        mut receiver2: Receiver<U>,
        tx: mpsc::UnboundedSender<(T, U)>,
    ) where
        U: Message + Send + 'static,
    {
        loop {
            let msg1 = receiver1.recv().await;
            let msg2 = receiver2.recv().await;

            match (msg1, msg2) {
                (Some(m1), Some(m2)) => {
                    if tx.send((m1, m2)).is_err() {
                        break;
                    }
                }
                _ => {
                    break;
                }
            }
        }
    }
}