mecha10_core/

stream.rs

//! Stream combinators for declarative message processing
//!
//! This module provides a fluent API for transforming and filtering message streams.
//!
//! # Examples
//!
//! ## Filtering
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::perception;
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Only process high-confidence detections
//! let mut detections = ctx.subscribe(perception::DETECTIONS)
//!     .filter(|d| d.iter().any(|det| det.confidence > 0.8))
//!     .await?;
//!
//! while let Some(dets) = detections.recv().await {
//!     // All detections have confidence > 0.8
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Throttling
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::sensor;
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Limit to 10 Hz
//! let mut images = ctx.subscribe(sensor::CAMERA_RGB)
//!     .throttle(Duration::from_millis(100))
//!     .await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Latest Only
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::sensor;
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Always get the latest, drop old messages
//! let mut odom = ctx.subscribe(sensor::ODOMETRY)
//!     .latest()
//!     .await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Zipping Streams
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::sensor;
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Combine camera images with LiDAR scans
//! let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
//! let mut scans = ctx.subscribe(sensor::LIDAR_SCAN).await?;
//!
//! let mut zipped = images.zip(scans);
//!
//! while let Some((image, scan)) = zipped.recv().await {
//!     // Process synchronized sensor data
//!     println!("Image: {}x{}, LiDAR: {} points",
//!         image.width, image.height, scan.ranges.len());
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Windowing
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::perception;
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Collect detections over 1-second windows for rate analysis
//! let mut windowed = ctx.subscribe(perception::DETECTIONS)
//!     .window(Duration::from_secs(1));
//!
//! while let Some(detections) = windowed.recv().await {
//!     let rate = detections.len() as f32; // detections per second
//!     println!("Detection rate: {} Hz", rate);
//!
//!     // Calculate average confidence over the window
//!     let avg_conf = detections.iter()
//!         .map(|d| d.confidence)
//!         .sum::<f32>() / detections.len() as f32;
//!     println!("Average confidence: {:.2}", avg_conf);
//! }
//! # Ok(())
//! # }
//! ```

use crate::context::Receiver;
use crate::messages::Message;
use std::time::{Duration, Instant};
use tokio::sync::mpsc;

/// Builder for creating filtered/transformed message streams
///
/// This builder allows chaining operations to create declarative
/// message processing pipelines.
pub struct StreamBuilder<T: Message> {
    receiver: Receiver<T>,
    #[allow(clippy::type_complexity)]
    filter_fn: Option<Box<dyn Fn(&T) -> bool + Send + Sync>>,
    #[allow(clippy::type_complexity)]
    map_fn: Option<Box<dyn Fn(T) -> T + Send + Sync>>,
    throttle_duration: Option<Duration>,
    debounce_duration: Option<Duration>,
    latest_only: bool,
    #[allow(dead_code)]
    batch_size: Option<usize>,
    #[allow(dead_code)]
    batch_timeout: Option<Duration>,
}

impl<T: Message + Send + 'static> StreamBuilder<T> {
    /// Create a new stream builder from a receiver
    pub fn new(receiver: Receiver<T>) -> Self {
        Self {
            receiver,
            filter_fn: None,
            map_fn: None,
            throttle_duration: None,
            debounce_duration: None,
            latest_only: false,
            batch_size: None,
            batch_timeout: None,
        }
    }

    /// Filter messages based on a predicate
    ///
    /// Only messages where the predicate returns `true` will be passed through.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(mut receiver: Receiver<Image>) -> Result<()> {
    /// let filtered = receiver
    ///     .filter(|img| img.width >= 640)
    ///     .build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn filter<F>(mut self, predicate: F) -> Self
    where
        F: Fn(&T) -> bool + Send + Sync + 'static,
    {
        self.filter_fn = Some(Box::new(predicate));
        self
    }

    /// Throttle messages to a maximum rate
    ///
    /// Messages will be rate-limited to at most one per `duration`.
    /// If multiple messages arrive during the throttle period, only the
    /// first one is kept and the rest are dropped.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(mut receiver: Receiver<Image>) -> Result<()> {
    /// // Max 10 Hz
    /// let throttled = receiver
    ///     .throttle(Duration::from_millis(100))
    ///     .build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn throttle(mut self, duration: Duration) -> Self {
        self.throttle_duration = Some(duration);
        self
    }

    /// Keep only the latest message, dropping older ones
    ///
    /// Useful for scenarios where you only care about the most recent
    /// state and processing old messages is wasteful.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(mut receiver: Receiver<Odometry>) -> Result<()> {
    /// let latest = receiver.latest().build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn latest(mut self) -> Self {
        self.latest_only = true;
        self
    }

    /// Transform messages using a mapping function
    ///
    /// Applies the given function to each message. The function must return
    /// the same type (for type-changing transformations, process messages manually).
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(mut receiver: Receiver<Image>) -> Result<()> {
    /// // Resize images to thumbnails (returns Image)
    /// let thumbnails = receiver
    ///     .map(|img| resize_to_thumbnail(img))
    ///     .build();
    ///
    /// // Normalize image brightness
    /// let normalized = receiver
    ///     .map(|mut img| {
    ///         normalize_brightness(&mut img);
    ///         img
    ///     })
    ///     .build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn map<F>(mut self, mapper: F) -> Self
    where
        F: Fn(T) -> T + Send + Sync + 'static,
    {
        self.map_fn = Some(Box::new(mapper));
        self
    }

    /// Debounce messages - ignore duplicates within a time window
    ///
    /// Only the first message is kept; subsequent messages within the debounce
    /// period are dropped. This is different from throttle - debounce resets
    /// the timer on each new message.
    ///
    /// Useful for:
    /// - Button press handling (ignore bounces)
    /// - Noisy sensor data
    /// - Rapid user input events
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(mut receiver: Receiver<ButtonPress>) -> Result<()> {
    /// // Ignore button bounces within 300ms
    /// let debounced = receiver
    ///     .debounce(Duration::from_millis(300))
    ///     .build();
    ///
    /// // Only get one event per 300ms window
    /// while let Some(press) = debounced.recv().await {
    ///     handle_button_press(press);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn debounce(mut self, duration: Duration) -> Self {
        self.debounce_duration = Some(duration);
        self
    }

    /// 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::prelude::*;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Zip camera images with LiDAR scans
    /// let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
    /// let mut scans = ctx.subscribe(sensor::LIDAR_SCAN).await?;
    ///
    /// let mut zipped = images.zip(scans).build();
    ///
    /// 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 {
            // Wait for messages from both streams
            let msg1 = receiver1.recv().await;
            let msg2 = receiver2.recv().await;

            match (msg1, msg2) {
                (Some(m1), Some(m2)) => {
                    // Both messages available, send tuple
                    if tx.send((m1, m2)).is_err() {
                        break; // Receiver dropped
                    }
                }
                _ => {
                    // One or both streams ended
                    break;
                }
            }
        }
    }

    /// 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::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;
                }
            }
        }
    }

    /// Build the transformed receiver
    ///
    /// Applies all the configured transformations and returns a new receiver.
    pub fn build(self) -> Receiver<T> {
        let (tx, rx) = mpsc::unbounded_channel();

        tokio::spawn(async move {
            self.run_pipeline(tx).await;
        });

        Receiver::from(rx)
    }

    /// Internal method to run the processing pipeline
    async fn run_pipeline(mut self, tx: mpsc::UnboundedSender<T>) {
        let mut last_send_time = None::<Instant>;
        let mut last_message_time = None::<Instant>;

        while let Some(mut msg) = self.receiver.recv().await {
            let now = Instant::now();

            // Apply filter
            if let Some(ref filter) = self.filter_fn {
                if !filter(&msg) {
                    continue;
                }
            }

            // Apply map transformation
            if let Some(ref mapper) = self.map_fn {
                msg = mapper(msg);
            }

            // Apply debounce - drop messages within debounce window
            if let Some(debounce_dur) = self.debounce_duration {
                if let Some(last_time) = last_message_time {
                    if now.duration_since(last_time) < debounce_dur {
                        // Message arrived too soon, drop it
                        last_message_time = Some(now); // Reset timer
                        continue;
                    }
                }
                last_message_time = Some(now);
            }

            // Apply throttle
            if let Some(throttle_dur) = self.throttle_duration {
                if let Some(last_time) = last_send_time {
                    let elapsed = last_time.elapsed();
                    if elapsed < throttle_dur {
                        continue;
                    }
                }
                last_send_time = Some(Instant::now());
            }

            // Handle latest-only mode
            if self.latest_only {
                // Drain any pending messages
                while self.receiver.try_recv().is_ok() {
                    // Discard old messages
                }
            }

            // Send the message
            if tx.send(msg).is_err() {
                // Receiver dropped, stop processing
                break;
            }
        }
    }
}

/// Builder for batched message streams
///
/// This builder collects messages into batches based on size or timeout,
/// whichever comes first. This solves the type transformation challenge by
/// using a separate builder type.
pub struct BatchBuilder<T: Message> {
    receiver: Receiver<T>,
    batch_size: usize,
    batch_timeout: Duration,
}

impl<T: Message + Send + Clone + 'static> BatchBuilder<T> {
    /// Create a new batch builder
    pub fn new(receiver: Receiver<T>, batch_size: usize, batch_timeout: Duration) -> Self {
        Self {
            receiver,
            batch_size,
            batch_timeout,
        }
    }

    /// Build the batched receiver
    ///
    /// Spawns a task that collects messages into batches and emits them
    /// when either the batch size is reached or the timeout expires.
    pub fn build(self) -> Receiver<Vec<T>>
    where
        T: Clone,
    {
        let (tx, rx) = mpsc::unbounded_channel();

        tokio::spawn(async move {
            self.run_batch_pipeline(tx).await;
        });

        Receiver::from(rx)
    }

    /// Internal method to run the batch collection pipeline
    async fn run_batch_pipeline(mut self, tx: mpsc::UnboundedSender<Vec<T>>) {
        let mut batch = Vec::with_capacity(self.batch_size);
        let mut deadline = tokio::time::Instant::now() + self.batch_timeout;

        loop {
            tokio::select! {
                // Receive next message
                msg_opt = self.receiver.recv() => {
                    match msg_opt {
                        Some(msg) => {
                            batch.push(msg);

                            // If batch is full, emit immediately
                            if batch.len() >= self.batch_size {
                                if tx.send(batch.clone()).is_err() {
                                    break; // Receiver dropped
                                }
                                batch.clear();
                                deadline = tokio::time::Instant::now() + self.batch_timeout;
                            }
                        }
                        None => {
                            // Stream ended - send remaining batch if any
                            if !batch.is_empty() {
                                let _ = tx.send(batch);
                            }
                            break;
                        }
                    }
                }

                // Batch timeout - emit collected messages
                _ = tokio::time::sleep_until(deadline) => {
                    if !batch.is_empty() {
                        if tx.send(batch.clone()).is_err() {
                            break; // Receiver dropped
                        }
                        batch.clear();
                    }
                    deadline = tokio::time::Instant::now() + self.batch_timeout;
                }
            }
        }
    }
}

/// Extension trait to add stream combinator methods to Receiver
pub trait ReceiverExt<T: Message>: Sized {
    /// Start building a stream pipeline
    fn stream(self) -> StreamBuilder<T>;

    /// Filter messages based on a predicate
    fn filter<F>(self, predicate: F) -> StreamBuilder<T>
    where
        F: Fn(&T) -> bool + Send + Sync + 'static;

    /// Transform messages using a mapping function
    fn map<F>(self, mapper: F) -> StreamBuilder<T>
    where
        F: Fn(T) -> T + Send + Sync + 'static;

    /// Debounce messages - ignore duplicates within a time window
    fn debounce(self, duration: Duration) -> StreamBuilder<T>;

    /// Throttle messages to a maximum rate
    fn throttle(self, duration: Duration) -> StreamBuilder<T>;

    /// Keep only the latest message
    fn latest(self) -> StreamBuilder<T>;

    /// Batch messages together
    ///
    /// Collects messages into batches of size `size`, or until `timeout`
    /// elapses, whichever comes first. Returns a BatchBuilder which can
    /// be built into a Receiver<Vec<T>>.
    ///
    /// # Arguments
    ///
    /// * `size` - Maximum number of messages per batch
    /// * `timeout` - Maximum time to wait before emitting a partial batch
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # use std::time::Duration;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Collect messages into batches of 10, or every 1 second
    /// let mut batched = ctx.subscribe(sensor::IMU)
    ///     .await?
    ///     .batch(10, Duration::from_secs(1))
    ///     .build();
    ///
    /// while let Some(batch) = batched.recv().await {
    ///     println!("Received batch of {} messages", batch.len());
    ///     // Process batch efficiently
    /// }
    /// # Ok(())
    /// # }
    /// ```
    fn batch(self, size: usize, timeout: Duration) -> BatchBuilder<T>
    where
        T: Clone;

    /// Zip this stream with another stream
    fn zip<U>(self, other: Receiver<U>) -> Receiver<(T, U)>
    where
        U: Message + Send + 'static;

    /// Window messages over a time period
    fn window(self, window_duration: Duration) -> Receiver<Vec<T>>
    where
        T: Clone;
}

impl<T: Message + Send + 'static> ReceiverExt<T> for Receiver<T> {
    fn stream(self) -> StreamBuilder<T> {
        StreamBuilder::new(self)
    }

    fn filter<F>(self, predicate: F) -> StreamBuilder<T>
    where
        F: Fn(&T) -> bool + Send + Sync + 'static,
    {
        StreamBuilder::new(self).filter(predicate)
    }

    fn map<F>(self, mapper: F) -> StreamBuilder<T>
    where
        F: Fn(T) -> T + Send + Sync + 'static,
    {
        StreamBuilder::new(self).map(mapper)
    }

    fn debounce(self, duration: Duration) -> StreamBuilder<T> {
        StreamBuilder::new(self).debounce(duration)
    }

    fn throttle(self, duration: Duration) -> StreamBuilder<T> {
        StreamBuilder::new(self).throttle(duration)
    }

    fn latest(self) -> StreamBuilder<T> {
        StreamBuilder::new(self).latest()
    }

    fn batch(self, size: usize, timeout: Duration) -> BatchBuilder<T>
    where
        T: Clone,
    {
        BatchBuilder::new(self, size, timeout)
    }

    fn zip<U>(self, other: Receiver<U>) -> Receiver<(T, U)>
    where
        U: Message + Send + 'static,
    {
        StreamBuilder::new(self).zip(other)
    }

    fn window(self, window_duration: Duration) -> Receiver<Vec<T>>
    where
        T: Clone,
    {
        StreamBuilder::new(self).window(window_duration)
    }
}