camel_api/
multicast.rs

1use crate::error::CamelError;
2use crate::exchange::Exchange;
3use std::sync::Arc;
4use std::time::Duration;
5
6pub type MulticastAggregationFn = Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync>;
7
8#[derive(Clone, Default)]
9pub enum MulticastStrategy {
10    #[default]
11    LastWins,
12    CollectAll,
13    Original,
14    Custom(MulticastAggregationFn),
15}
16
17impl std::fmt::Debug for MulticastStrategy {
18    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
19        match self {
20            MulticastStrategy::LastWins => f.write_str("LastWins"),
21            MulticastStrategy::CollectAll => f.write_str("CollectAll"),
22            MulticastStrategy::Original => f.write_str("Original"),
23            MulticastStrategy::Custom(_) => f.write_str("Custom(..)"),
24        }
25    }
26}
27
28#[derive(Clone)]
29pub struct MulticastConfig {
30    pub parallel: bool,
31    pub parallel_limit: Option<usize>,
32    pub stop_on_exception: bool,
33    pub timeout: Option<Duration>,
34    pub aggregation: MulticastStrategy,
35}
36
37impl std::fmt::Debug for MulticastConfig {
38    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
39        f.debug_struct("MulticastConfig")
40            .field("parallel", &self.parallel)
41            .field("parallel_limit", &self.parallel_limit)
42            .field("stop_on_exception", &self.stop_on_exception)
43            .field("timeout", &self.timeout)
44            .field("aggregation", &self.aggregation)
45            .finish()
46    }
47}
48
49impl MulticastConfig {
50    pub fn new() -> Self {
51        Self {
52            parallel: false,
53            parallel_limit: None,
54            stop_on_exception: false,
55            timeout: None,
56            aggregation: MulticastStrategy::default(),
57        }
58    }
59
60    pub fn parallel(mut self, parallel: bool) -> Self {
61        self.parallel = parallel;
62        self
63    }
64
65    pub fn parallel_limit(mut self, limit: usize) -> Self {
66        self.parallel_limit = Some(limit);
67        self
68    }
69
70    pub fn stop_on_exception(mut self, stop: bool) -> Self {
71        self.stop_on_exception = stop;
72        self
73    }
74
75    pub fn timeout(mut self, duration: Duration) -> Self {
76        self.timeout = Some(duration);
77        self
78    }
79
80    pub fn aggregation(mut self, strategy: MulticastStrategy) -> Self {
81        self.aggregation = strategy;
82        self
83    }
84
85    /// Validates the configuration.
86    ///
87    /// Returns `Err(CamelError::Config)` if `parallel_limit` is set to 0,
88    /// which would cause a `Semaphore::new(0)` panic at runtime.
89    pub fn validate(&self) -> Result<(), CamelError> {
90        if self.parallel && self.parallel_limit == Some(0) {
91            return Err(CamelError::Config(
92                "multicast parallel_limit must be > 0".to_string(),
93            ));
94        }
95        Ok(())
96    }
97}
98
99impl Default for MulticastConfig {
100    fn default() -> Self {
101        Self::new()
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108    use std::time::Duration;
109
110    #[test]
111    fn test_multicast_config_defaults() {
112        let config = MulticastConfig::new();
113        assert!(!config.parallel);
114        assert!(config.parallel_limit.is_none());
115        assert!(!config.stop_on_exception);
116        assert!(config.timeout.is_none());
117        assert!(matches!(config.aggregation, MulticastStrategy::LastWins));
118    }
119
120    #[test]
121    fn test_multicast_config_builder() {
122        let config = MulticastConfig::new()
123            .parallel(true)
124            .parallel_limit(4)
125            .stop_on_exception(true)
126            .timeout(Duration::from_millis(500));
127
128        assert!(config.parallel);
129        assert_eq!(config.parallel_limit, Some(4));
130        assert!(config.stop_on_exception);
131        assert_eq!(config.timeout, Some(Duration::from_millis(500)));
132    }
133}
camel_api/multicast.rs

camel_api/
multicast.rs
