mecha10-core 0.6.5

//! `StreamBuilder::build`: running the configured filter/map/throttle/debounce/latest pipeline.

use super::StreamBuilder;
use crate::context::Receiver;
use crate::messages::Message;
use std::time::Instant;
use tokio::sync::mpsc;

impl<T: Message + Send + 'static> StreamBuilder<T> {
    /// 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();

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

            if let Some(ref mapper) = self.map_fn {
                msg = mapper(msg);
            }

            if let Some(debounce_dur) = self.debounce_duration {
                if let Some(last_time) = last_message_time {
                    if now.duration_since(last_time) < debounce_dur {
                        last_message_time = Some(now);
                        continue;
                    }
                }
                last_message_time = Some(now);
            }

            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());
            }

            if self.latest_only {
                while self.receiver.try_recv().is_ok() {}
            }

            if tx.send(msg).is_err() {
                break;
            }
        }
    }
}