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

camel_api/
multicast.rs
