mecha10-core 0.6.3

//! `StreamBuilder`'s data and pipeline configuration methods.

use crate::context::Receiver;
use crate::messages::Message;
use std::time::Duration;

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

    crate::builder_setter!(
        /// 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_core::prelude::*;
        /// # async fn example(mut receiver: Receiver<Image>) -> Result<()> {
        /// // Max 10 Hz
        /// let throttled = receiver
        ///     .throttle(Duration::from_millis(100))
        ///     .build();
        /// # Ok(())
        /// # }
        /// ```
        into_some, throttle, throttle_duration, Duration
    );

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

    crate::builder_setter!(
        /// 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_core::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(())
        /// # }
        /// ```
        into_some, debounce, debounce_duration, Duration
    );
}