mecha10-core 0.9.0

//! `debounce` and `window` time-based combinators.
//!
//! duplicate: false positive — cargo-dupes groups `debounce`/`window` (here) with
//! `Receiver::spawn_relay` (relay.rs) as an exact duplicate. The part that's genuinely shared
//! across all three — creating an output channel matching `self`'s boundedness, spawning a
//! background task, and wrapping the result back into a `Receiver` — is already factored out
//! into `matching_channel` and used by all of them (see relay.rs). What's left in `debounce`/
//! `window` is a `tokio::select!` loop reacting to both incoming messages *and* a timer, which
//! is structurally unlike `spawn_relay`'s per-item-only loop and unlike each other in its
//! actual state machine (quiet-period debounce vs. tumbling-window batching). Forcing those two
//! into one generic time-combinator would add a real risk of subtly changing the concurrency
//! behavior of both for a purely mechanical dedup — not a change to make outside a dedicated
//! review of this module.

use super::relay::matching_channel;
use super::types::Receiver;
use std::time::Duration;
use tokio::time::{sleep_until, Instant};

impl<T> Receiver<T> {
    /// 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_core::prelude::*;
    /// # use mecha10_core::topics::{actuator, sensor};
    /// # 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 velocity command updates
    /// let mut stable_status = ctx.subscribe(actuator::CMD_VEL)
    ///     .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) = matching_channel::<T, T>(&self.inner);

        let mut receiver = self;
        tokio::spawn(async move {
            let mut deadline: Option<Instant> = None;
            let mut pending_msg: Option<T> = None;
            loop {
                tokio::select! {
                    msg = receiver.recv() => {
                        match msg {
                            Some(new_msg) => {
                                pending_msg = Some(new_msg);
                                deadline = Some(Instant::now() + duration);
                            }
                            None => {
                                if let Some(msg_to_send) = pending_msg.take() {
                                    let _ = tx.send(msg_to_send).await;
                                }
                                break;
                            }
                        }
                    }

                    _ = sleep_until(deadline.unwrap_or_else(Instant::now)), if deadline.is_some() => {
                        if let Some(msg_to_send) = pending_msg.take() {
                            if tx.send(msg_to_send).await.is_err() {
                                break;
                            }
                        }

                        deadline = None;
                    }
                }
            }
        });

        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_core::prelude::*;
    /// # use mecha10_core::messages::Image;
    /// # use mecha10_core::topics::sensor;
    /// # 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.linear_acceleration[0]).sum::<f32>() / batch.len() as f32;
    ///         println!("Average accel: {} (n={})", avg, batch.len());
    ///     }
    /// }
    ///
    /// // Count messages per second
    /// let mut counts = ctx.subscribe(sensor::BATTERY_VOLTAGE)
    ///     .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(sensor::CAMERA_RGB)
    ///     .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(())
    /// # }
    /// # async fn process_batch(batch: &[Image]) -> Result<()> { 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) = matching_channel::<T, Vec<T>>(&self.inner);

        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! {
                    msg = receiver.recv() => {
                        match msg {
                            Some(new_msg) => {
                                buffer.push(new_msg);
                            }
                            None => {
                                if !buffer.is_empty() {
                                    let _ = tx.send(buffer).await;
                                }
                                break;
                            }
                        }
                    }

                    _ = interval.tick() => {
                        let batch = std::mem::take(&mut buffer);
                        if tx.send(batch).await.is_err() {
                            break;
                        }
                    }
                }
            }
        });

        Receiver { inner: rx }
    }
}