mecha10_core/context/

receiver.rs

//! Receiver for subscribing to messages
//!
//! Provides a type-safe wrapper around tokio channels for receiving messages
//! from subscribed topics with support for both bounded and unbounded channels.
//!
//! # Backpressure Strategies
//!
//! The receiver supports multiple backpressure strategies to prevent memory exhaustion
//! from fast publishers. Choose based on your use case:
//!
//! ## 1. Unbounded (Default)
//! No backpressure - messages queue indefinitely. Use for low-rate topics.
//!
//! ```rust,no_run
//! # use mecha10::prelude::*;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::CAMERA_RGB).await?;
//! // Unlimited queue - risk of memory growth
//! # Ok(())
//! # }
//! ```
//!
//! ## 2. Bounded with Blocking (`.with_capacity()`)
//! Publisher waits when buffer full. Use when all messages are important.
//!
//! ```rust,no_run
//! # use mecha10::prelude::*;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::CAMERA_RGB)
//!     .await?
//!     .with_capacity(50); // Block publisher when 50 messages queued
//! # Ok(())
//! # }
//! ```
//!
//! ## 3. Bounded with Drop Oldest (`.with_drop_oldest()`)
//! Drops old messages when buffer full. Use for real-time sensor data.
//!
//! ```rust,no_run
//! # use mecha10::prelude::*;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::LIDAR_SCAN)
//!     .await?
//!     .with_drop_oldest(10); // Keep only 10 newest scans
//! # Ok(())
//! # }
//! ```
//!
//! ## 4. Rate Limiting (`.debounce()`)
//! Limit emission rate. Use to reduce update frequency.
//!
//! ```rust,no_run
//! # use mecha10::prelude::*;
//! # use std::time::Duration;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::CAMERA_RGB)
//!     .await?
//!     .debounce(Duration::from_millis(100)); // Max 10 Hz
//! # Ok(())
//! # }
//! ```
//!
//! ## Stream Combinators
//!
//! Transform and combine streams with powerful combinators:
//!
//! ```rust,no_run
//! # use mecha10::prelude::*;
//! # use std::time::Duration;
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Transform messages
//! let mut scaled = ctx.subscribe(sensor::CAMERA_RGB)
//!     .await?
//!     .map(|img| scale_down(&img));
//!
//! // Filter messages
//! let mut valid = ctx.subscribe(sensor::LIDAR)
//!     .await?
//!     .filter(|scan| scan.range > 0.1);
//!
//! // Combine streams (sensor fusion)
//! let camera = ctx.subscribe(sensor::CAMERA_RGB).await?;
//! let lidar = ctx.subscribe(sensor::LIDAR).await?;
//! let mut fused = camera.zip(lidar);
//!
//! // Window messages over time (for aggregation)
//! let mut windowed = ctx.subscribe(sensor::IMU)
//!     .await?
//!     .window(Duration::from_millis(100))      // Collect 100ms batches
//!     .map(|batch| calculate_average(&batch)); // Aggregate
//!
//! // Chain combinators
//! let mut processed = ctx.subscribe(sensor::CAMERA_RGB)
//!     .await?
//!     .with_drop_oldest(20)                    // Buffer 20 latest
//!     .debounce(Duration::from_millis(100))    // Max 10 Hz
//!     .filter(|img| img.width == 640)          // Only 640px wide
//!     .map(|img| preprocess(&img));            // Transform
//! # Ok(())
//! # }
//! ```
//!
//! # Monitoring Backpressure
//!
//! Use `.len()`, `.capacity()`, and `.utilization()` to monitor queue health:
//!
//! ```rust,no_run
//! # use mecha10::prelude::*;
//! # async fn example(rx: &mut Receiver<String>) {
//! if let Some(util) = rx.utilization() {
//!     if util > 0.8 {
//!         println!("WARNING: Queue is {}% full!", util * 100.0);
//!     }
//! }
//! # }
//! ```

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, Instant};

/// Internal channel type for the receiver
pub(crate) enum ReceiverInner<T> {
    /// Unbounded channel (no backpressure, risk of memory exhaustion)
    Unbounded(mpsc::UnboundedReceiver<T>),
    /// Bounded channel (with backpressure, prevents memory exhaustion)
    Bounded(mpsc::Receiver<T>),
}

/// Receiver for subscribing to messages
///
/// This is a wrapper around tokio channels that provides a unified interface
/// for both bounded and unbounded channels. Bounded channels provide backpressure
/// to prevent memory exhaustion from fast publishers.
///
/// # Backpressure
///
/// When using bounded channels (via `with_capacity()`), the receiver applies
/// backpressure when the channel is full. This prevents memory exhaustion but
/// may cause publishers to block or drop messages depending on the strategy.
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// # async fn example(ctx: &Context) -> Result<()> {
/// // Unbounded receiver (default, no backpressure)
/// let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
///
/// // Bounded receiver with capacity limit (applies backpressure)
/// let mut images = ctx.subscribe(sensor::CAMERA_RGB)
///     .await?
///     .with_capacity(100);  // Hold max 100 messages
///
/// while let Some(image) = images.recv().await {
///     // Process image
/// }
/// # Ok(())
/// # }
/// ```
pub struct Receiver<T> {
    pub(crate) inner: ReceiverInner<T>,
}

impl<T> Receiver<T> {
    /// Receive the next message
    ///
    /// Returns `None` when all senders have been dropped.
    pub async fn recv(&mut self) -> Option<T> {
        match &mut self.inner {
            ReceiverInner::Unbounded(rx) => rx.recv().await,
            ReceiverInner::Bounded(rx) => rx.recv().await,
        }
    }

    /// Try to receive a message without blocking
    ///
    /// Returns an error if the channel is empty or all senders have been dropped.
    pub fn try_recv(&mut self) -> std::result::Result<T, mpsc::error::TryRecvError> {
        match &mut self.inner {
            ReceiverInner::Unbounded(rx) => rx.try_recv(),
            ReceiverInner::Bounded(rx) => rx.try_recv(),
        }
    }

    /// Check if the channel is currently empty
    ///
    /// Note: This is a point-in-time check and may change immediately after.
    pub fn is_empty(&self) -> bool {
        match &self.inner {
            ReceiverInner::Unbounded(rx) => rx.is_empty(),
            ReceiverInner::Bounded(rx) => rx.is_empty(),
        }
    }

    /// Get the number of messages currently in the channel
    ///
    /// This is useful for monitoring backpressure and queue depth.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(mut rx: Receiver<String>) {
    /// let depth = rx.len();
    /// if depth > 50 {
    ///     warn!("Channel backlog growing: {} messages", depth);
    /// }
    /// # }
    /// ```
    pub fn len(&self) -> usize {
        match &self.inner {
            ReceiverInner::Unbounded(rx) => rx.len(),
            ReceiverInner::Bounded(rx) => rx.len(),
        }
    }

    /// Convert an unbounded receiver to a bounded one with the specified capacity
    ///
    /// This creates a new bounded channel and spawns a task to forward messages.
    /// Messages are dropped if the new channel fills up, applying backpressure.
    ///
    /// # Arguments
    ///
    /// * `capacity` - Maximum number of messages to buffer
    ///
    /// # Backpressure Strategy
    ///
    /// When the bounded channel is full, the oldest message is dropped to make
    /// room for new messages. This "drop oldest" strategy ensures recent data
    /// is always available.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let mut images = ctx.subscribe(sensor::CAMERA_RGB)
    ///     .await?
    ///     .with_capacity(50);  // Limit to 50 images in queue
    ///
    /// // If publisher sends faster than we process, old images are dropped
    /// while let Some(image) = images.recv().await {
    ///     // Always get relatively recent images
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_capacity(self, capacity: usize) -> Self
    where
        T: 'static + Send,
    {
        match self.inner {
            ReceiverInner::Bounded(_) => {
                // Already bounded, no conversion needed
                self
            }
            ReceiverInner::Unbounded(mut unbounded_rx) => {
                let (tx, rx) = mpsc::channel(capacity);

                // Spawn task to forward messages with backpressure
                tokio::spawn(async move {
                    while let Some(msg) = unbounded_rx.recv().await {
                        // Try to send with backpressure
                        // If channel is full, this will wait until space is available
                        if tx.send(msg).await.is_err() {
                            // Receiver dropped, stop forwarding
                            break;
                        }
                    }
                });

                Self {
                    inner: ReceiverInner::Bounded(rx),
                }
            }
        }
    }

    /// Convert an unbounded receiver to a bounded one that drops oldest messages
    ///
    /// Similar to `with_capacity()`, but explicitly uses a "drop oldest" strategy
    /// when the buffer is full. This is useful for real-time data where the most
    /// recent value is more important than historical data.
    ///
    /// # Arguments
    ///
    /// * `capacity` - Maximum number of messages to buffer
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // For sensor data, we want the latest reading
    /// let mut lidar = ctx.subscribe(sensor::LIDAR_SCAN)
    ///     .await?
    ///     .with_drop_oldest(10);  // Keep only 10 most recent scans
    ///
    /// while let Some(scan) = lidar.recv().await {
    ///     // Always process recent data
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_drop_oldest(self, capacity: usize) -> Self
    where
        T: 'static + Send,
    {
        match self.inner {
            ReceiverInner::Bounded(_) => self,
            ReceiverInner::Unbounded(mut unbounded_rx) => {
                let (tx, rx) = mpsc::channel(capacity);

                tokio::spawn(async move {
                    while let Some(msg) = unbounded_rx.recv().await {
                        // Try to send without waiting
                        if tx.try_send(msg).is_err() {
                            // Channel full or closed
                            // For try_send on bounded channel, this means it's full
                            // Drop the message and continue (drop oldest strategy)
                            continue;
                        }
                    }
                });

                Self {
                    inner: ReceiverInner::Bounded(rx),
                }
            }
        }
    }

    /// Get a reference to the maximum capacity if this is a bounded channel
    ///
    /// Returns `None` for unbounded channels or the capacity for bounded ones.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(rx: Receiver<String>) {
    /// if let Some(cap) = rx.capacity() {
    ///     println!("Channel capacity: {}", cap);
    ///     println!("Current usage: {}/{}", rx.len(), cap);
    ///     println!("Utilization: {:.1}%", (rx.len() as f32 / cap as f32) * 100.0);
    /// } else {
    ///     println!("Unbounded channel, {} messages queued", rx.len());
    /// }
    /// # }
    /// ```
    pub fn capacity(&self) -> Option<usize> {
        match &self.inner {
            ReceiverInner::Unbounded(_) => None,
            ReceiverInner::Bounded(rx) => Some(rx.max_capacity()),
        }
    }

    /// Get the current utilization ratio (0.0 to 1.0) for bounded channels
    ///
    /// Returns `None` for unbounded channels.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(rx: Receiver<String>) {
    /// if let Some(utilization) = rx.utilization() {
    ///     if utilization > 0.8 {
    ///         warn!("Channel is {}% full, backpressure may occur", utilization * 100.0);
    ///     }
    /// }
    /// # }
    /// ```
    pub fn utilization(&self) -> Option<f32> {
        self.capacity()
            .map(|cap| if cap == 0 { 1.0 } else { self.len() as f32 / cap as f32 })
    }

    /// Map (transform) messages as they are received
    ///
    /// Creates a new receiver that applies a transformation function to each message.
    /// This is useful for data transformations like resizing images, converting units,
    /// or extracting fields from complex messages.
    ///
    /// # Arguments
    ///
    /// * `f` - Transformation function to apply to each message
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Resize camera images on the fly
    /// let mut resized_images = ctx.subscribe(sensor::CAMERA_RGB)
    ///     .await?
    ///     .map(|img| resize_image(img, 320, 240));
    ///
    /// while let Some(small_image) = resized_images.recv().await {
    ///     // Process resized image
    /// }
    ///
    /// // Extract specific fields from complex messages
    /// let mut speeds = ctx.subscribe(motor::STATUS)
    ///     .await?
    ///     .map(|status| status.current_speed);
    ///
    /// while let Some(speed) = speeds.recv().await {
    ///     println!("Current speed: {}", speed);
    /// }
    /// # Ok(())
    /// # }
    /// # fn resize_image(img: String, w: u32, h: u32) -> String { img }
    /// ```
    pub fn map<U, F>(self, f: F) -> Receiver<U>
    where
        F: FnMut(T) -> U + Send + 'static,
        T: Send + 'static,
        U: Send + 'static,
    {
        let (tx, rx) = match &self.inner {
            ReceiverInner::Unbounded(_) => {
                let (tx, rx) = mpsc::unbounded_channel();
                (MappedSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
            }
            ReceiverInner::Bounded(bounded_rx) => {
                let capacity = bounded_rx.max_capacity();
                let (tx, rx) = mpsc::channel(capacity);
                (MappedSender::Bounded(tx), ReceiverInner::Bounded(rx))
            }
        };

        // Spawn task to apply transformation
        let mut receiver = self;
        let mut transform = f;
        tokio::spawn(async move {
            while let Some(msg) = receiver.recv().await {
                let transformed = transform(msg);

                // Send transformed message
                let send_result = match tx {
                    MappedSender::Unbounded(ref tx) => tx.send(transformed).map_err(|_| ()),
                    MappedSender::Bounded(ref tx) => tx.send(transformed).await.map_err(|_| ()),
                };

                if send_result.is_err() {
                    // Receiver dropped, stop processing
                    break;
                }
            }
        });

        Receiver { inner: rx }
    }

    /// Filter messages based on a predicate
    ///
    /// Creates a new receiver that only passes through messages that satisfy the predicate.
    /// This is useful for filtering sensor data, ignoring invalid readings, or implementing
    /// conditional processing.
    ///
    /// # Arguments
    ///
    /// * `predicate` - Function that returns `true` for messages to keep, `false` to drop
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Only process high-speed motor data
    /// let mut high_speeds = ctx.subscribe(motor::STATUS)
    ///     .await?
    ///     .filter(|status| status.speed > 0.5);
    ///
    /// while let Some(status) = high_speeds.recv().await {
    ///     println!("High speed detected: {}", status.speed);
    /// }
    ///
    /// // Filter out invalid lidar readings
    /// let mut valid_scans = ctx.subscribe(sensor::LIDAR_SCAN)
    ///     .await?
    ///     .filter(|scan| scan.ranges.iter().all(|&r| r > 0.0 && r < 100.0));
    ///
    /// while let Some(scan) = valid_scans.recv().await {
    ///     // All ranges are valid
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn filter<F>(self, mut predicate: F) -> Receiver<T>
    where
        F: FnMut(&T) -> bool + Send + 'static,
        T: Send + 'static,
    {
        let (tx, rx) = match &self.inner {
            ReceiverInner::Unbounded(_) => {
                let (tx, rx) = mpsc::unbounded_channel();
                (FilteredSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
            }
            ReceiverInner::Bounded(bounded_rx) => {
                let capacity = bounded_rx.max_capacity();
                let (tx, rx) = mpsc::channel(capacity);
                (FilteredSender::Bounded(tx), ReceiverInner::Bounded(rx))
            }
        };

        // Spawn task to filter messages
        let mut receiver = self;
        tokio::spawn(async move {
            while let Some(msg) = receiver.recv().await {
                // Apply predicate
                if !predicate(&msg) {
                    continue; // Skip messages that don't match
                }

                // Send filtered message
                let send_result = match &tx {
                    FilteredSender::Unbounded(tx) => tx.send(msg).map_err(|_| ()),
                    FilteredSender::Bounded(tx) => tx.send(msg).await.map_err(|_| ()),
                };

                if send_result.is_err() {
                    // Receiver dropped, stop processing
                    break;
                }
            }
        });

        Receiver { inner: rx }
    }

    /// Filter and map messages in one operation
    ///
    /// Combines filtering and mapping into a single efficient operation. The function
    /// returns `Some(U)` for messages to keep (and transform), or `None` to drop.
    /// This is more efficient than chaining `.filter()` and `.map()` separately.
    ///
    /// # Arguments
    ///
    /// * `f` - Function that returns `Some(transformed)` to keep, `None` to drop
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Extract speed only when robot is moving
    /// let mut active_speeds = ctx.subscribe(motor::STATUS)
    ///     .await?
    ///     .filter_map(|status| {
    ///         if status.speed > 0.1 {
    ///             Some(status.speed)
    ///         } else {
    ///             None
    ///         }
    ///     });
    ///
    /// while let Some(speed) = active_speeds.recv().await {
    ///     println!("Robot moving at: {}", speed);
    /// }
    ///
    /// // Parse valid sensor readings
    /// let mut parsed_data = ctx.subscribe(sensor::RAW_DATA)
    ///     .await?
    ///     .filter_map(|raw| parse_sensor_data(&raw).ok());
    ///
    /// while let Some(data) = parsed_data.recv().await {
    ///     // Process successfully parsed data
    /// }
    /// # Ok(())
    /// # }
    /// # fn parse_sensor_data(raw: &String) -> Result<String> { Ok(raw.clone()) }
    /// ```
    pub fn filter_map<U, F>(self, mut f: F) -> Receiver<U>
    where
        F: FnMut(T) -> Option<U> + Send + 'static,
        T: Send + 'static,
        U: Send + 'static,
    {
        let (tx, rx) = match &self.inner {
            ReceiverInner::Unbounded(_) => {
                let (tx, rx) = mpsc::unbounded_channel();
                (FilterMapSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
            }
            ReceiverInner::Bounded(bounded_rx) => {
                let capacity = bounded_rx.max_capacity();
                let (tx, rx) = mpsc::channel(capacity);
                (FilterMapSender::Bounded(tx), ReceiverInner::Bounded(rx))
            }
        };

        // Spawn task to filter and map
        let mut receiver = self;
        tokio::spawn(async move {
            while let Some(msg) = receiver.recv().await {
                // Apply filter_map function
                if let Some(transformed) = f(msg) {
                    // Send transformed message
                    let send_result = match &tx {
                        FilterMapSender::Unbounded(tx) => tx.send(transformed).map_err(|_| ()),
                        FilterMapSender::Bounded(tx) => tx.send(transformed).await.map_err(|_| ()),
                    };

                    if send_result.is_err() {
                        // Receiver dropped, stop processing
                        break;
                    }
                }
                // If None, message is dropped (filtered out)
            }
        });

        Receiver { inner: rx }
    }

    /// Debounce (rate-limit) messages to prevent rapid bursts
    ///
    /// Creates a new receiver that only emits a message if a certain duration has passed
    /// since the last emission. This is useful for reducing update frequency of high-rate
    /// sensors or preventing UI update storms.
    ///
    /// **Strategy:** "Emit latest after quiet period"
    /// - Messages are buffered during the debounce period
    /// - Only the most recent message is emitted after the period expires
    /// - Earlier messages within the period are dropped
    ///
    /// # Arguments
    ///
    /// * `duration` - Minimum time between emissions
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # use std::time::Duration;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Reduce camera updates to max 10 Hz (100ms debounce)
    /// let mut debounced_images = ctx.subscribe(sensor::CAMERA_RGB)
    ///     .await?
    ///     .debounce(Duration::from_millis(100));
    ///
    /// while let Some(image) = debounced_images.recv().await {
    ///     // Process at most 10 images/sec
    /// }
    ///
    /// // Prevent rapid motor status updates
    /// let mut stable_status = ctx.subscribe(motor::STATUS)
    ///     .await?
    ///     .debounce(Duration::from_millis(50));
    ///
    /// while let Some(status) = stable_status.recv().await {
    ///     // Only get updates every 50ms
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn debounce(self, duration: Duration) -> Receiver<T>
    where
        T: Send + 'static,
    {
        let (tx, rx) = match &self.inner {
            ReceiverInner::Unbounded(_) => {
                let (tx, rx) = mpsc::unbounded_channel();
                (DebouncedSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
            }
            ReceiverInner::Bounded(bounded_rx) => {
                let capacity = bounded_rx.max_capacity();
                let (tx, rx) = mpsc::channel(capacity);
                (DebouncedSender::Bounded(tx), ReceiverInner::Bounded(rx))
            }
        };

        // Spawn task to debounce messages
        let mut receiver = self;
        tokio::spawn(async move {
            let mut last_emit = Instant::now();
            let mut pending_msg: Option<T> = None;

            loop {
                // Calculate time until next allowed emission
                let elapsed = last_emit.elapsed();
                let remaining = if elapsed >= duration {
                    Duration::from_millis(0)
                } else {
                    duration - elapsed
                };

                // Wait for either a message or debounce timeout
                tokio::select! {
                    // New message received
                    msg = receiver.recv() => {
                        match msg {
                            Some(new_msg) => {
                                // Store as pending (replaces any previous pending)
                                pending_msg = Some(new_msg);

                                // If debounce period passed, emit immediately
                                if remaining.is_zero() {
                                    if let Some(msg_to_send) = pending_msg.take() {
                                        let send_result = match &tx {
                                            DebouncedSender::Unbounded(tx) => tx.send(msg_to_send).map_err(|_| ()),
                                            DebouncedSender::Bounded(tx) => tx.send(msg_to_send).await.map_err(|_| ()),
                                        };

                                        if send_result.is_err() {
                                            break; // Receiver dropped
                                        }

                                        last_emit = Instant::now();
                                    }
                                }
                                // Otherwise, message stays pending until timeout
                            }
                            None => {
                                // Source closed, send any pending message and exit
                                if let Some(msg_to_send) = pending_msg.take() {
                                    let _ = match &tx {
                                        DebouncedSender::Unbounded(tx) => tx.send(msg_to_send).map_err(|_| ()),
                                        DebouncedSender::Bounded(tx) => tx.send(msg_to_send).await.map_err(|_| ()),
                                    };
                                }
                                break;
                            }
                        }
                    }

                    // Debounce timeout - emit pending message if any
                    _ = sleep(remaining), if pending_msg.is_some() && !remaining.is_zero() => {
                        if let Some(msg_to_send) = pending_msg.take() {
                            let send_result = match &tx {
                                DebouncedSender::Unbounded(tx) => tx.send(msg_to_send).map_err(|_| ()),
                                DebouncedSender::Bounded(tx) => tx.send(msg_to_send).await.map_err(|_| ()),
                            };

                            if send_result.is_err() {
                                break; // Receiver dropped
                            }

                            last_emit = Instant::now();
                        }
                    }
                }
            }
        });

        Receiver { inner: rx }
    }

    /// Window messages over a time period
    ///
    /// Creates a new receiver that collects messages over a fixed time window and emits
    /// them as a batch (Vec) at the end of each window. This is useful for:
    /// - Aggregating sensor readings (averaging, min/max)
    /// - Batch processing for efficiency
    /// - Time-series analysis
    /// - Rate calculation (count messages per window)
    ///
    /// # Behavior
    ///
    /// - **Fixed Windows**: Each window has a fixed duration
    /// - **Non-overlapping**: Windows don't overlap (tumbling window)
    /// - **Batch Emission**: All messages in window emitted as Vec at window end
    /// - **Empty Windows**: Empty Vec emitted if no messages received
    /// - **Termination**: Emits final partial window when source ends
    ///
    /// # Arguments
    ///
    /// * `window_duration` - Duration of each time window
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # use std::time::Duration;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Average IMU readings over 100ms windows
    /// let mut windowed = ctx.subscribe(sensor::IMU)
    ///     .await?
    ///     .window(Duration::from_millis(100));
    ///
    /// while let Some(batch) = windowed.recv().await {
    ///     if !batch.is_empty() {
    ///         let avg = batch.iter().map(|imu| imu.accel_x).sum::<f32>() / batch.len() as f32;
    ///         println!("Average accel: {} (n={})", avg, batch.len());
    ///     }
    /// }
    ///
    /// // Count messages per second
    /// let mut counts = ctx.subscribe("/events")
    ///     .await?
    ///     .window(Duration::from_secs(1))
    ///     .map(|batch| batch.len());
    ///
    /// while let Some(count) = counts.recv().await {
    ///     println!("Events/sec: {}", count);
    /// }
    ///
    /// // Batch process for efficiency
    /// let mut batches = ctx.subscribe("/data")
    ///     .await?
    ///     .window(Duration::from_millis(50));
    ///
    /// while let Some(batch) = batches.recv().await {
    ///     // Process entire batch at once (more efficient than one-by-one)
    ///     process_batch(&batch).await?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Performance
    ///
    /// - **Memory**: Buffers all messages within window (use reasonable window sizes)
    /// - **Latency**: Adds latency equal to window duration
    /// - **CPU**: Minimal overhead for window management
    ///
    /// # Caveats
    ///
    /// - Window size affects memory usage (smaller windows = less memory)
    /// - All messages are delivered at window end (batch latency)
    /// - For sliding windows, consider multiple overlapping `.window()` subscriptions
    pub fn window(self, window_duration: Duration) -> Receiver<Vec<T>>
    where
        T: Send + 'static,
    {
        let (tx, rx) = match &self.inner {
            ReceiverInner::Unbounded(_) => {
                let (tx, rx) = mpsc::unbounded_channel();
                (WindowedSender::Unbounded(tx), ReceiverInner::Unbounded(rx))
            }
            ReceiverInner::Bounded(bounded_rx) => {
                let capacity = bounded_rx.max_capacity();
                let (tx, rx) = mpsc::channel(capacity);
                (WindowedSender::Bounded(tx), ReceiverInner::Bounded(rx))
            }
        };

        // Spawn task to window messages
        let mut receiver = self;
        tokio::spawn(async move {
            let mut buffer: Vec<T> = Vec::new();
            let mut interval = tokio::time::interval(window_duration);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

            loop {
                tokio::select! {
                    // New message received
                    msg = receiver.recv() => {
                        match msg {
                            Some(new_msg) => {
                                buffer.push(new_msg);
                            }
                            None => {
                                // Source ended - emit final window and exit
                                if !buffer.is_empty() {
                                    let _ = match &tx {
                                        WindowedSender::Unbounded(tx) => tx.send(buffer).map_err(|_| ()),
                                        WindowedSender::Bounded(tx) => tx.send(buffer).await.map_err(|_| ()),
                                    };
                                }
                                break;
                            }
                        }
                    }

                    // Window duration elapsed
                    _ = interval.tick() => {
                        // Emit current window (even if empty)
                        let batch = std::mem::take(&mut buffer);
                        let send_result = match &tx {
                            WindowedSender::Unbounded(tx) => tx.send(batch).map_err(|_| ()),
                            WindowedSender::Bounded(tx) => tx.send(batch).await.map_err(|_| ()),
                        };

                        if send_result.is_err() {
                            // Receiver dropped, stop windowing
                            break;
                        }
                    }
                }
            }
        });

        Receiver { inner: rx }
    }

    /// 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
        let capacity = match (&self.inner, &other.inner) {
            (ReceiverInner::Unbounded(_), ReceiverInner::Unbounded(_)) => None,
            (ReceiverInner::Bounded(a), ReceiverInner::Unbounded(_)) => Some(a.max_capacity()),
            (ReceiverInner::Unbounded(_), ReceiverInner::Bounded(b)) => Some(b.max_capacity()),
            (ReceiverInner::Bounded(a), ReceiverInner::Bounded(b)) => Some(a.max_capacity().min(b.max_capacity())),
        };

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

// Helper enums for sender types in combinators
enum MappedSender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

enum FilteredSender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

enum FilterMapSender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

enum DebouncedSender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

enum ZippedSender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

enum WindowedSender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

impl<T> From<mpsc::UnboundedReceiver<T>> for Receiver<T> {
    fn from(inner: mpsc::UnboundedReceiver<T>) -> Self {
        Self {
            inner: ReceiverInner::Unbounded(inner),
        }
    }
}

impl<T> From<mpsc::Receiver<T>> for Receiver<T> {
    fn from(inner: mpsc::Receiver<T>) -> Self {
        Self {
            inner: ReceiverInner::Bounded(inner),
        }
    }
}