mecha10-core 0.6.5

//! Aggregated Topics - Multi-topic subscription with pattern matching
//!
//! This module provides functionality for subscribing to multiple topics simultaneously
//! using glob patterns. Messages from all matching topics are multiplexed into a single
//! stream, with each message tagged by its source topic path.
//!
//! # Use Cases
//!
//! - Dashboard monitoring (subscribe to all cameras at once)
//! - Data recording (capture all sensors)
//! - Multi-sensor fusion (combine multiple IMU sources)
//! - Telemetry aggregation (all system metrics)
//!
//! # Example
//!
//! ```rust,no_run
//! use mecha10_core::prelude::*;
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Subscribe to all camera topics
//! let mut cameras = ctx.subscribe_aggregated::<ImageMessage>("/sensor/camera/*").await?;
//!
//! while let Some((topic, image)) = cameras.recv().await {
//!     println!("Received from {}: {}x{}", topic, image.width, image.height);
//! }
//! # Ok(())
//! # }
//! ```

use crate::messages::Message;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

/// Representation of receiver types for aggregation
///
/// Supports both unbounded and bounded receivers for flexible backpressure strategies.
pub enum ReceiverType<T> {
    /// Unbounded channel receiver (no backpressure)
    Unbounded(mpsc::UnboundedReceiver<T>),
    /// Bounded channel receiver (with backpressure)
    Bounded(mpsc::Receiver<T>),
}

/// Multi-topic receiver that tags messages with their source topic
///
/// Messages from all subscribed topics are multiplexed into a single stream.
/// Each message includes the topic path it originated from.
///
/// Supports both unbounded and bounded channels with fair scheduling.
///
/// # Type Parameters
///
/// * `T` - The message type that all subscribed topics must share
///
/// # Examples
///
/// ```rust,no_run
/// # use mecha10_core::prelude::*;
/// # async fn example(ctx: &Context) -> Result<()> {
/// let mut cameras = ctx.subscribe_aggregated::<ImageMessage>("/sensor/camera/*").await?;
///
/// // Receive messages from any camera
/// while let Some((topic, image)) = cameras.recv().await {
///     let camera_name = topic.rsplit('/').next().unwrap_or("unknown");
///     println!("Camera {}: {}x{}", camera_name, image.width, image.height);
/// }
/// # Ok(())
/// # }
/// ```
pub struct AggregatedReceiver<T: Message> {
    /// The glob pattern used for this aggregated receiver
    pattern: String,

    /// List of subscribed topics
    topics: Vec<String>,

    /// Internal channel for receiving multiplexed messages
    rx: mpsc::UnboundedReceiver<(String, T)>,

    /// Background task handles for forwarding messages
    _tasks: Vec<JoinHandle<()>>,
}

impl<T: Message> AggregatedReceiver<T> {
    /// Create a new aggregated receiver
    ///
    /// # Arguments
    ///
    /// * `pattern` - The glob pattern used to match topics
    /// * `receivers` - List of (topic_path, receiver) tuples for each matching topic
    ///
    /// # Returns
    ///
    /// A new `AggregatedReceiver` that multiplexes all input receivers
    pub fn new(pattern: String, receivers: Vec<(String, mpsc::UnboundedReceiver<T>)>) -> Self {
        let receivers = receivers
            .into_iter()
            .map(|(topic_path, receiver)| (topic_path, ReceiverType::Unbounded(receiver)))
            .collect();

        Self::new_mixed(pattern, receivers)
    }

    /// Create a new aggregated receiver supporting both bounded and unbounded channels
    ///
    /// This version spawns one task per receiver, naturally providing fair scheduling
    /// without busy-waiting. Each task forwards messages with topic tags to the
    /// aggregated channel.
    ///
    /// # Arguments
    ///
    /// * `pattern` - The glob pattern used to match topics
    /// * `receivers` - List of (topic_path, receiver) tuples for each matching topic
    ///
    /// # Returns
    ///
    /// A new `AggregatedReceiver` that multiplexes all input receivers with backpressure support
    ///
    /// # Performance
    ///
    /// Spawns one task per receiver. Each task efficiently waits on its receiver using
    /// async `.recv()`, providing optimal performance without polling or busy-waiting.
    /// Tokio's scheduler ensures fair multiplexing across all tasks.
    pub fn new_mixed(pattern: String, receivers: Vec<(String, ReceiverType<T>)>) -> Self {
        let (tx, rx) = mpsc::unbounded_channel();
        let mut tasks = Vec::new();
        let mut topics = Vec::new();

        for (topic_path, mut receiver) in receivers {
            topics.push(topic_path.clone());
            let tx = tx.clone();
            let topic_path_clone = topic_path.clone();

            let task = tokio::spawn(async move {
                loop {
                    let msg_opt = match &mut receiver {
                        ReceiverType::Unbounded(rx) => rx.recv().await,
                        ReceiverType::Bounded(rx) => rx.recv().await,
                    };

                    match msg_opt {
                        Some(message) => {
                            if tx.send((topic_path_clone.clone(), message)).is_err() {
                                break;
                            }
                        }
                        None => {
                            break;
                        }
                    }
                }
            });

            tasks.push(task);
        }

        Self {
            pattern,
            topics,
            rx,
            _tasks: tasks,
        }
    }

    /// Receive the next message from any subscribed topic
    ///
    /// Returns a tuple of `(topic_path, message)`.
    /// Returns `None` when all senders are closed.
    ///
    /// # Returns
    ///
    /// * `Some((topic, message))` - Next message with its source topic
    /// * `None` - All subscriptions have been closed
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// # use mecha10_core::prelude::*;
    /// # async fn example(mut cameras: AggregatedReceiver<ImageMessage>) -> Result<()> {
    /// while let Some((topic, image)) = cameras.recv().await {
    ///     println!("Received from {}: {}x{}", topic, image.width, image.height);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn recv(&mut self) -> Option<(String, T)> {
        self.rx.recv().await
    }

    /// Get the glob pattern for this aggregated receiver
    ///
    /// # Returns
    ///
    /// The pattern string used to match topics
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// # use mecha10_core::aggregated::AggregatedReceiver;
    /// # use mecha10_core::types::Message;
    /// # fn example<T: Message>(receiver: &AggregatedReceiver<T>) {
    /// let pattern = receiver.pattern();
    /// println!("Monitoring pattern: {}", pattern);
    /// # }
    /// ```
    pub fn pattern(&self) -> &str {
        &self.pattern
    }

    /// Get list of currently subscribed topics
    ///
    /// # Returns
    ///
    /// A vector of topic paths that match the pattern
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// # use mecha10_core::aggregated::AggregatedReceiver;
    /// # use mecha10_core::types::Message;
    /// # fn example<T: Message>(receiver: &AggregatedReceiver<T>) {
    /// let topics = receiver.topics();
    /// println!("Subscribed to {} topics: {:?}", topics.len(), topics);
    /// # }
    /// ```
    pub fn topics(&self) -> &[String] {
        &self.topics
    }

    /// Get the number of subscribed topics
    ///
    /// # Returns
    ///
    /// The count of topics currently being aggregated
    pub fn topic_count(&self) -> usize {
        self.topics.len()
    }
}

impl<T: Message> Unpin for AggregatedReceiver<T> {}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn new_forwards_messages_from_unbounded_receivers() {
        let (tx_a, rx_a) = mpsc::unbounded_channel::<i32>();
        let (tx_b, rx_b) = mpsc::unbounded_channel::<i32>();

        let mut aggregated = AggregatedReceiver::new(
            "topic/*".to_string(),
            vec![("topic/a".to_string(), rx_a), ("topic/b".to_string(), rx_b)],
        );

        assert_eq!(aggregated.pattern(), "topic/*");
        assert_eq!(aggregated.topic_count(), 2);

        tx_a.send(1).unwrap();
        tx_b.send(2).unwrap();
        drop(tx_a);
        drop(tx_b);

        let mut received = vec![aggregated.recv().await.unwrap(), aggregated.recv().await.unwrap()];
        received.sort();
        assert_eq!(received, vec![("topic/a".to_string(), 1), ("topic/b".to_string(), 2)]);

        assert_eq!(aggregated.recv().await, None);
    }
}

/// Pattern matching utilities for topic glob patterns
pub mod pattern {
    /// Check if a topic path matches a glob pattern
    ///
    /// Supports full glob syntax via the `globset` crate:
    /// - `*` matches any sequence of characters within a path segment
    /// - `**` matches any sequence of path segments (including none)
    /// - `{a,b}` matches either a or b
    /// - `?` matches any single character
    /// - `[abc]` matches any character in the set
    /// - `[!abc]` matches any character not in the set
    ///
    /// # Arguments
    ///
    /// * `path` - The topic path to test
    /// * `pattern` - The glob pattern to match against
    ///
    /// # Returns
    ///
    /// `true` if the path matches the pattern, `false` otherwise
    ///
    /// # Examples
    ///
    /// ```rust
    /// use mecha10_core::aggregated::pattern::matches_pattern;
    ///
    /// assert!(matches_pattern("/sensor/camera/rgb", "/sensor/camera/*"));
    /// assert!(matches_pattern("/sensor/camera/depth", "/sensor/camera/*"));
    /// assert!(!matches_pattern("/sensor/lidar/scan", "/sensor/camera/*"));
    ///
    /// assert!(matches_pattern("/sensor/camera/rgb", "/sensor/**"));
    /// assert!(matches_pattern("/sensor/lidar/scan", "/sensor/**"));
    ///
    /// // Advanced patterns
    /// assert!(matches_pattern("/sensor/camera1/rgb", "/sensor/camera?/rgb"));
    /// assert!(matches_pattern("/sensor/camera/left", "/sensor/camera/{left,right}"));
    /// assert!(matches_pattern("/sensor/imu0/data", "/sensor/imu[0-9]/data"));
    /// ```
    pub fn matches_pattern(path: &str, pattern: &str) -> bool {
        use globset::GlobBuilder;

        let glob = GlobBuilder::new(pattern)
            .literal_separator(false)
            .backslash_escape(true)
            .build();

        match glob {
            Ok(glob_pattern) => {
                let matcher = glob_pattern.compile_matcher();
                matcher.is_match(path)
            }
            Err(_) => path == pattern,
        }
    }

    /// Convert a glob pattern to a Redis KEYS pattern
    ///
    /// # Arguments
    ///
    /// * `pattern` - The glob pattern
    ///
    /// # Returns
    ///
    /// A Redis-compatible KEYS pattern
    ///
    /// # Examples
    ///
    /// ```rust
    /// use mecha10_core::aggregated::pattern::convert_glob_to_redis;
    ///
    /// assert_eq!(
    ///     convert_glob_to_redis("/sensor/camera/*"),
    ///     "mecha10:/sensor/camera/*"
    /// );
    /// ```
    pub fn convert_glob_to_redis(pattern: &str) -> String {
        format!("mecha10:{}", pattern)
    }
}