camel-api 0.24.0

Core traits and interfaces for rust-camel
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use crate::error::CamelError;
use crate::exchange::Exchange;
use std::sync::Arc;
use std::time::Duration;

pub type MulticastAggregationFn = Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync>;

#[derive(Clone, Default)]
pub enum MulticastStrategy {
    #[default]
    LastWins,
    CollectAll,
    Original,
    Custom(MulticastAggregationFn),
}

impl std::fmt::Debug for MulticastStrategy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            MulticastStrategy::LastWins => f.write_str("LastWins"),
            MulticastStrategy::CollectAll => f.write_str("CollectAll"),
            MulticastStrategy::Original => f.write_str("Original"),
            MulticastStrategy::Custom(_) => f.write_str("Custom(..)"),
        }
    }
}

#[derive(Clone)]
pub struct MulticastConfig {
    pub parallel: bool,
    pub parallel_limit: Option<usize>,
    pub stop_on_exception: bool,
    pub timeout: Option<Duration>,
    pub aggregation: MulticastStrategy,
}

impl std::fmt::Debug for MulticastConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MulticastConfig")
            .field("parallel", &self.parallel)
            .field("parallel_limit", &self.parallel_limit)
            .field("stop_on_exception", &self.stop_on_exception)
            .field("timeout", &self.timeout)
            .field("aggregation", &self.aggregation)
            .finish()
    }
}

impl MulticastConfig {
    pub fn new() -> Self {
        Self {
            parallel: false,
            parallel_limit: None,
            stop_on_exception: false,
            timeout: None,
            aggregation: MulticastStrategy::default(),
        }
    }

    pub fn parallel(mut self, parallel: bool) -> Self {
        self.parallel = parallel;
        self
    }

    pub fn parallel_limit(mut self, limit: usize) -> Self {
        self.parallel_limit = Some(limit);
        self
    }

    pub fn stop_on_exception(mut self, stop: bool) -> Self {
        self.stop_on_exception = stop;
        self
    }

    pub fn timeout(mut self, duration: Duration) -> Self {
        self.timeout = Some(duration);
        self
    }

    pub fn aggregation(mut self, strategy: MulticastStrategy) -> Self {
        self.aggregation = strategy;
        self
    }

    /// Validates the configuration.
    ///
    /// Returns `Err(CamelError::Config)` if `parallel_limit` is set to 0,
    /// which would cause a `Semaphore::new(0)` panic at runtime.
    pub fn validate(&self) -> Result<(), CamelError> {
        if self.parallel && self.parallel_limit == Some(0) {
            return Err(CamelError::Config(
                "multicast parallel_limit must be > 0".to_string(),
            ));
        }
        Ok(())
    }
}

impl Default for MulticastConfig {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn test_multicast_config_defaults() {
        let config = MulticastConfig::new();
        assert!(!config.parallel);
        assert!(config.parallel_limit.is_none());
        assert!(!config.stop_on_exception);
        assert!(config.timeout.is_none());
        assert!(matches!(config.aggregation, MulticastStrategy::LastWins));
    }

    #[test]
    fn test_multicast_config_builder() {
        let config = MulticastConfig::new()
            .parallel(true)
            .parallel_limit(4)
            .stop_on_exception(true)
            .timeout(Duration::from_millis(500));

        assert!(config.parallel);
        assert_eq!(config.parallel_limit, Some(4));
        assert!(config.stop_on_exception);
        assert_eq!(config.timeout, Some(Duration::from_millis(500)));
    }
}