mecha10-core 0.8.0

//! `ReceiverExt`: the fluent stream-combinator entry point on `Receiver<T>`.

use super::{BatchBuilder, StreamBuilder};
use crate::context::Receiver;
use crate::messages::Message;
use std::time::Duration;

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_core::prelude::*;
    /// # use mecha10_core::topics::sensor;
    /// # 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)
    }
}