mecha10_core/

publish.rs

//! Conditional publishing for declarative message publishing
//!
//! This module provides a fluent API for conditional and throttled publishing.
//!
//! # Examples
//!
//! ## Conditional Publishing
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::perception;
//!
//! # async fn example(ctx: &Context, detections: Vec<Detection>) -> Result<()> {
//! // Only publish when detections are non-empty
//! ctx.publish_to(perception::DETECTIONS, &detections)
//!     .when(|d| !d.is_empty())
//!     .await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Throttled Publishing
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topics::system;
//!
//! # async fn example(ctx: &Context, status: String) -> Result<()> {
//! // Publish at most once per second
//! ctx.publish_to(system::NODE_STATUS, &status)
//!     .throttle(Duration::from_secs(1))
//!     .await?;
//! # Ok(())
//! # }
//! ```

use crate::context::Context;
use crate::error::Result;
use crate::messages::Message;
use crate::topics::Topic;
use serde::Serialize;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;

/// Global throttle state for tracking last publish times
static THROTTLE_STATE: once_cell::sync::Lazy<Arc<Mutex<HashMap<String, Instant>>>> =
    once_cell::sync::Lazy::new(|| Arc::new(Mutex::new(HashMap::new())));

/// Builder for conditional/throttled publishing
///
/// This builder allows chaining operations to create declarative
/// publishing behavior.
pub struct PublishBuilder<'a, 'b, T: Message + Serialize> {
    ctx: &'a Context,
    topic: Topic<T>,
    message: &'b T,
    #[allow(clippy::type_complexity)]
    condition: Option<Box<dyn Fn(&T) -> bool + Send + Sync>>,
    throttle_duration: Option<Duration>,
}

impl<'a, 'b, T: Message + Serialize> PublishBuilder<'a, 'b, T> {
    /// Create a new publish builder
    pub fn new(ctx: &'a Context, topic: Topic<T>, message: &'b T) -> Self {
        Self {
            ctx,
            topic,
            message,
            condition: None,
            throttle_duration: None,
        }
    }

    /// Only publish if condition is true
    ///
    /// If the predicate returns `false`, the message will not be published.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context, detections: Vec<Detection>) -> Result<()> {
    /// ctx.publish_to(perception::DETECTIONS, &detections)
    ///     .when(|d| !d.is_empty())
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn when<F>(mut self, predicate: F) -> Self
    where
        F: Fn(&T) -> bool + Send + Sync + 'static,
    {
        self.condition = Some(Box::new(predicate));
        self
    }

    /// Throttle publishing to a maximum rate
    ///
    /// Messages will be rate-limited to at most one per `duration`.
    /// If you try to publish more frequently, the message will be dropped.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10::prelude::*;
    /// # async fn example(ctx: &Context, status: String) -> Result<()> {
    /// // Max once per second
    /// ctx.publish_to(system::NODE_STATUS, &status)
    ///     .throttle(Duration::from_secs(1))
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn throttle(mut self, duration: Duration) -> Self {
        self.throttle_duration = Some(duration);
        self
    }

    /// Execute the publish with all configured conditions
    ///
    /// This method applies all the configured filters and throttling,
    /// then publishes the message if all conditions are met.
    pub async fn execute(self) -> Result<()> {
        // Apply condition check
        if let Some(ref condition) = self.condition {
            if !condition(self.message) {
                // Condition not met, skip publishing
                return Ok(());
            }
        }

        // Apply throttle check
        if let Some(throttle_dur) = self.throttle_duration {
            let topic_key = format!("{}-{}", self.ctx.node_id(), self.topic.path());

            let mut throttle_state = THROTTLE_STATE.lock().await;

            if let Some(last_time) = throttle_state.get(&topic_key) {
                let elapsed = last_time.elapsed();
                if elapsed < throttle_dur {
                    // Throttle limit not reached, skip publishing
                    return Ok(());
                }
            }

            // Update last publish time
            throttle_state.insert(topic_key, Instant::now());
        }

        // All conditions met, publish the message
        self.ctx.publish_to(self.topic, self.message).await
    }
}

/// Extension trait to add conditional publishing to Context
pub trait ContextPublishExt {
    /// Start building a conditional publish operation
    fn publish_with<'a, 'b, T>(&'a self, topic: Topic<T>, message: &'b T) -> PublishBuilder<'a, 'b, T>
    where
        T: Message + Serialize;
}

impl ContextPublishExt for Context {
    fn publish_with<'a, 'b, T>(&'a self, topic: Topic<T>, message: &'b T) -> PublishBuilder<'a, 'b, T>
    where
        T: Message + Serialize,
    {
        PublishBuilder::new(self, topic, message)
    }
}

// Allow PublishBuilder to be awaited directly
impl<'a, 'b, T: Message + Serialize> std::future::IntoFuture for PublishBuilder<'a, 'b, T>
where
    'b: 'a,
{
    type Output = Result<()>;
    type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(self.execute())
    }
}