camel_processor/
multicast.rs

1//! ## Stop semantics (ADR-0025)
2//!
3//! This segment implements `OutcomePipeline` and propagates `PipelineOutcome::Stopped(ex)` with the exchange state intact (including mutations made inside the segment body before Stop fired). See ADR-0025 §3 (stopped-exchange-state-preservation invariant).
4
5use std::future::Future;
6use std::pin::Pin;
7use std::task::{Context, Poll};
8
9use tower::Service;
10
11use camel_api::{
12    Body, BoxProcessor, CamelError, Exchange, MulticastConfig, MulticastStrategy, Value,
13};
14
15// ── Metadata property keys ─────────────────────────────────────────────
16
17/// Property key for the zero-based index of the endpoint being invoked.
18pub const CAMEL_MULTICAST_INDEX: &str = "CamelMulticastIndex";
19/// Property key indicating whether this is the last endpoint invocation.
20pub const CAMEL_MULTICAST_COMPLETE: &str = "CamelMulticastComplete";
21
22// ── MulticastService ───────────────────────────────────────────────────
23
24/// Tower Service implementing the Multicast EIP.
25///
26/// **DoS bound (R3-L4):** fan-out is bounded by the static endpoint list
27/// (operator-configured at route build, not attacker-controlled). In parallel
28/// mode, `parallel_limit` defaults to `endpoints.len()` so the code path always
29/// carries an explicit concurrency bound.
30///
31/// Sends a message to multiple endpoints, processing each independently,
32/// and then aggregating the results.
33///
34/// Supports both sequential and parallel processing modes, configurable
35/// via [`MulticastConfig::parallel`]. When parallel mode is enabled,
36/// all endpoints are invoked concurrently with optional concurrency
37/// limiting via [`MulticastConfig::parallel_limit`].
38#[derive(Clone)]
39pub struct MulticastService {
40    endpoints: Vec<BoxProcessor>,
41    config: MulticastConfig,
42}
43
44impl MulticastService {
45    /// Create a new `MulticastService` from a list of endpoints and a [`MulticastConfig`].
46    pub fn new(
47        endpoints: Vec<BoxProcessor>,
48        mut config: MulticastConfig,
49    ) -> Result<Self, CamelError> {
50        // R3-L4: parallel fan-out is bounded by the static endpoint list
51        // (operator-controlled at route build time, not attacker-controlled).
52        // Normalize parallel_limit to the explicit endpoint count so the
53        // parallel path always carries a concrete semaphore bound.
54        if config.parallel && config.parallel_limit.is_none() {
55            config.parallel_limit = Some(endpoints.len());
56        }
57        config.validate()?;
58        Ok(Self { endpoints, config })
59    }
60
61    /// Borrow the effective config (post-normalization).
62    pub fn config(&self) -> &MulticastConfig {
63        &self.config
64    }
65}
66
67impl Service<Exchange> for MulticastService {
68    type Response = Exchange;
69    type Error = CamelError;
70    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
71
72    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
73        // Do NOT aggregate endpoint readiness here.
74        // Each endpoint's readiness is checked per-endpoint inside
75        // process_sequential / process_parallel where stop_on_exception
76        // is respected. Fail-fast here would bypass that logic entirely.
77        Poll::Ready(Ok(()))
78    }
79
80    fn call(&mut self, exchange: Exchange) -> Self::Future {
81        let original = exchange.clone();
82        let endpoints = self.endpoints.clone();
83        let config = self.config.clone();
84
85        Box::pin(async move {
86            // If no endpoints, return original exchange unchanged
87            if endpoints.is_empty() {
88                return Ok(original);
89            }
90
91            let total = endpoints.len();
92
93            let results = if config.parallel {
94                // Process endpoints in parallel
95                process_parallel(exchange, endpoints, config.parallel_limit, total).await
96            } else {
97                // Process each endpoint sequentially
98                process_sequential(exchange, endpoints, config.stop_on_exception, total).await
99            };
100
101            // Aggregate results per strategy
102            aggregate(results, original, config.aggregation)
103        })
104    }
105}
106
107// ── Sequential processing ──────────────────────────────────────────────
108
109async fn process_sequential(
110    exchange: Exchange,
111    endpoints: Vec<BoxProcessor>,
112    stop_on_exception: bool,
113    total: usize,
114) -> Vec<Result<Exchange, CamelError>> {
115    let mut results = Vec::with_capacity(endpoints.len());
116
117    for (i, endpoint) in endpoints.into_iter().enumerate() {
118        // Clone the exchange for each endpoint
119        let mut cloned_exchange = exchange.clone();
120
121        // Set multicast metadata properties
122        cloned_exchange.set_property(CAMEL_MULTICAST_INDEX, Value::from(i as i64));
123        cloned_exchange.set_property(CAMEL_MULTICAST_COMPLETE, Value::Bool(i == total - 1));
124
125        let mut endpoint = endpoint;
126        match tower::ServiceExt::ready(&mut endpoint).await {
127            Err(e) => {
128                results.push(Err(e));
129                if stop_on_exception {
130                    break;
131                }
132            }
133            Ok(svc) => {
134                let result = svc.call(cloned_exchange).await;
135                let is_err = result.is_err();
136                results.push(result);
137                if stop_on_exception && is_err {
138                    break;
139                }
140            }
141        }
142    }
143
144    results
145}
146
147// ── Parallel processing ────────────────────────────────────────────────
148
149async fn process_parallel(
150    exchange: Exchange,
151    endpoints: Vec<BoxProcessor>,
152    parallel_limit: Option<usize>,
153    total: usize,
154) -> Vec<Result<Exchange, CamelError>> {
155    use std::sync::Arc;
156    use tokio::sync::Semaphore;
157
158    let semaphore = parallel_limit.map(|limit| Arc::new(Semaphore::new(limit)));
159
160    // Build futures for each endpoint
161    let futures: Vec<_> = endpoints
162        .into_iter()
163        .enumerate()
164        .map(|(i, mut endpoint)| {
165            let mut ex = exchange.clone();
166            ex.set_property(CAMEL_MULTICAST_INDEX, Value::from(i as i64));
167            ex.set_property(CAMEL_MULTICAST_COMPLETE, Value::Bool(i == total - 1));
168            let sem = semaphore.clone();
169            async move {
170                // Acquire semaphore permit if limit is set
171                let _permit = match &sem {
172                    Some(s) => match s.acquire().await {
173                        Ok(p) => Some(p),
174                        Err(_) => {
175                            return Err(CamelError::ProcessorError("semaphore closed".to_string()));
176                        }
177                    },
178                    None => None,
179                };
180
181                // Readiness errors propagate via `?` into the per-endpoint result;
182                // join_all ensures all endpoints run independently (no early abort).
183                tower::ServiceExt::ready(&mut endpoint).await?;
184                endpoint.call(ex).await
185            }
186        })
187        .collect();
188
189    // Execute all futures concurrently and collect results
190    futures::future::join_all(futures).await
191}
192
193// ── Aggregation ────────────────────────────────────────────────────────
194
195fn aggregate(
196    results: Vec<Result<Exchange, CamelError>>,
197    original: Exchange,
198    strategy: MulticastStrategy,
199) -> Result<Exchange, CamelError> {
200    match strategy {
201        MulticastStrategy::LastWins => {
202            // Return the last result (error or success).
203            // If last result is Err and stop_on_exception=false, return that error.
204            results.into_iter().last().unwrap_or_else(|| Ok(original))
205        }
206        MulticastStrategy::CollectAll => {
207            // Collect all bodies into a JSON array. Errors propagate.
208            let mut bodies = Vec::new();
209            for result in results {
210                let ex = result?;
211                let value = match &ex.input.body {
212                    Body::Text(s) => Value::String(s.clone()),
213                    Body::Json(v) => v.clone(),
214                    Body::Xml(s) => Value::String(s.clone()),
215                    Body::Bytes(b) => Value::String(String::from_utf8_lossy(b).into_owned()),
216                    Body::Empty => Value::Null,
217                    Body::Stream(s) => serde_json::json!({
218                        "_stream": {
219                            "origin": s.metadata.origin,
220                            "placeholder": true,
221                            "hint": "Materialize exchange body with .into_bytes() before multicast aggregation"
222                        }
223                    }),
224                };
225                bodies.push(value);
226            }
227            let mut out = original;
228            out.input.body = Body::Json(Value::Array(bodies));
229            Ok(out)
230        }
231        MulticastStrategy::Original => Ok(original),
232        MulticastStrategy::Custom(fold_fn) => {
233            // Fold using the custom function, starting from the first result.
234            let mut iter = results.into_iter();
235            let first = iter.next().unwrap_or_else(|| Ok(original.clone()))?;
236            iter.try_fold(first, |acc, next_result| {
237                let next = next_result?;
238                Ok(fold_fn(acc, next))
239            })
240        }
241    }
242}
243
244#[cfg(test)]
245#[path = "multicast_tests.rs"]
246mod tests;
camel_processor/multicast.rs

camel_processor/
multicast.rs
