camel_processor/
loop_eip.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use tower::{Service, ServiceExt};
6
7use camel_api::loop_eip::{LoopConfig, LoopMode, MAX_LOOP_ITERATIONS};
8use camel_api::{BoxProcessor, CamelError, Exchange, Value};
9
10pub const CAMEL_LOOP_INDEX: &str = "CamelLoopIndex";
11pub const CAMEL_LOOP_SIZE: &str = "CamelLoopSize";
12
13#[derive(Clone)]
14pub struct LoopService {
15    config: LoopConfig,
16    sub_pipeline: BoxProcessor,
17}
18
19impl LoopService {
20    pub fn new(config: LoopConfig, sub_pipeline: BoxProcessor) -> Self {
21        Self {
22            config,
23            sub_pipeline,
24        }
25    }
26}
27
28impl Service<Exchange> for LoopService {
29    type Response = Exchange;
30    type Error = CamelError;
31    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
32
33    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
34        Poll::Ready(Ok(()))
35    }
36
37    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
38        let config = self.config.clone();
39        let mut pipeline = self.sub_pipeline.clone();
40
41        Box::pin(async move {
42            match config.mode {
43                LoopMode::Count(n) => {
44                    // D-M9 Batch 1: clamp to MAX_LOOP_ITERATIONS. The audit
45                    // finding was that Count(n) had no cap (only While did);
46                    // a `count: u32::MAX` would exhaust CPU. The clamp is
47                    // applied uniformly to Count and While. The CAMEL_LOOP_SIZE
48                    // property is set to the *clamped* count so downstream
49                    // steps observe the effective iteration count.
50                    let n_clamped = n.min(MAX_LOOP_ITERATIONS);
51                    if n > MAX_LOOP_ITERATIONS {
52                        tracing::warn!(
53                            requested = n,
54                            clamped_to = MAX_LOOP_ITERATIONS,
55                            "LoopMode::Count exceeded MAX_LOOP_ITERATIONS; clamping"
56                        );
57                    }
58                    exchange.set_property(CAMEL_LOOP_SIZE, Value::from(n_clamped as u64));
59                    for i in 0..n_clamped {
60                        exchange.set_property(CAMEL_LOOP_INDEX, Value::from(i as u64));
61                        exchange = pipeline.ready().await?.call(exchange).await?;
62                    }
63                }
64                LoopMode::While(ref predicate) => {
65                    exchange.set_property(CAMEL_LOOP_SIZE, Value::from(0u64));
66                    for i in 0..MAX_LOOP_ITERATIONS {
67                        if !predicate(&exchange) {
68                            break;
69                        }
70                        exchange.set_property(CAMEL_LOOP_INDEX, Value::from(i as u64));
71                        exchange = pipeline.ready().await?.call(exchange).await?;
72                    }
73                    if predicate(&exchange) {
74                        tracing::warn!(
75                            "Loop while-mode hit MAX_LOOP_ITERATIONS ({}) safety guard. Predicate still true.",
76                            MAX_LOOP_ITERATIONS
77                        );
78                    }
79                }
80            }
81            Ok(exchange)
82        })
83    }
84}
85
86// ── LoopSegment (ADR-0025 OutcomePipeline) ─────────────────────────────
87
88/// Outcome-aware structural EIP segment for the Loop pattern.
89///
90/// Operates at the `PipelineOutcome` layer so that `Stopped(ex)` from a
91/// sub-step (e.g. Stop EIP) is preserved with the exchange including all
92/// mutations. Supports both Count and While modes, mirroring `LoopService`
93/// semantics exactly.
94///
95/// Unlike `LoopService` (which operates at the Tower layer), `LoopSegment`
96/// correctly short-circuits on `PipelineOutcome::Stopped` or `Failed`.
97pub struct LoopSegment {
98    pub config: camel_api::loop_eip::LoopConfig,
99    pub body: camel_api::OutcomeSegment,
100}
101
102impl Clone for LoopSegment {
103    fn clone(&self) -> Self {
104        Self {
105            config: self.config.clone(),
106            body: self.body.clone(),
107        }
108    }
109}
110
111impl camel_api::OutcomePipeline for LoopSegment {
112    fn clone_box(&self) -> Box<dyn camel_api::OutcomePipeline> {
113        Box::new(self.clone())
114    }
115
116    fn run<'a>(
117        &'a mut self,
118        exchange: camel_api::Exchange,
119    ) -> Pin<Box<dyn Future<Output = camel_api::PipelineOutcome> + Send + 'a>> {
120        use camel_api::loop_eip::MAX_LOOP_ITERATIONS;
121        use camel_api::{PipelineOutcome, Value};
122
123        let config = self.config.clone();
124        let body = &mut self.body;
125
126        Box::pin(async move {
127            match config.mode {
128                camel_api::loop_eip::LoopMode::Count(n) => {
129                    let n_clamped = n.min(MAX_LOOP_ITERATIONS);
130                    if n > MAX_LOOP_ITERATIONS {
131                        tracing::warn!(
132                            requested = n,
133                            clamped_to = MAX_LOOP_ITERATIONS,
134                            "LoopMode::Count exceeded MAX_LOOP_ITERATIONS; clamping"
135                        );
136                    }
137                    let mut ex = exchange;
138                    ex.set_property(CAMEL_LOOP_SIZE, Value::from(n_clamped as u64));
139                    for i in 0..n_clamped {
140                        ex.set_property(CAMEL_LOOP_INDEX, Value::from(i as u64));
141                        match body.run(ex).await {
142                            PipelineOutcome::Completed(next) => {
143                                ex = next;
144                            }
145                            other => return other,
146                        }
147                    }
148                    PipelineOutcome::Completed(ex)
149                }
150                camel_api::loop_eip::LoopMode::While(ref predicate) => {
151                    let mut ex = exchange;
152                    ex.set_property(CAMEL_LOOP_SIZE, Value::from(0u64));
153                    let mut i = 0u64;
154                    while i < MAX_LOOP_ITERATIONS as u64 {
155                        if !predicate(&ex) {
156                            break;
157                        }
158                        ex.set_property(CAMEL_LOOP_INDEX, Value::from(i));
159                        match body.run(ex).await {
160                            PipelineOutcome::Completed(next) => {
161                                ex = next;
162                            }
163                            other => return other,
164                        }
165                        i += 1;
166                    }
167                    if predicate(&ex) {
168                        tracing::warn!(
169                            "Loop while-mode hit MAX_LOOP_ITERATIONS ({}) safety guard. Predicate still true.",
170                            MAX_LOOP_ITERATIONS
171                        );
172                    }
173                    PipelineOutcome::Completed(ex)
174                }
175            }
176        })
177    }
178}
179
180#[cfg(test)]
181mod tests {
182    use std::sync::atomic::{AtomicUsize, Ordering};
183    use std::sync::{Arc, Mutex};
184
185    use camel_api::loop_eip::{LoopConfig, LoopMode, MAX_LOOP_ITERATIONS};
186    use camel_api::{
187        Body, BoxProcessor, BoxProcessorExt, CamelError, Exchange, IdentityProcessor, Message,
188    };
189    use tower::{Service, ServiceExt};
190
191    use super::{CAMEL_LOOP_INDEX, CAMEL_LOOP_SIZE, LoopService};
192
193    fn identity_pipeline() -> BoxProcessor {
194        BoxProcessor::new(IdentityProcessor)
195    }
196
197    fn counter_pipeline(counter: Arc<AtomicUsize>) -> BoxProcessor {
198        BoxProcessor::from_fn(move |exchange: Exchange| {
199            let counter = Arc::clone(&counter);
200            Box::pin(async move {
201                counter.fetch_add(1, Ordering::SeqCst);
202                Ok(exchange)
203            })
204        })
205    }
206
207    #[tokio::test]
208    async fn test_loop_count_iterates_n_times() {
209        let counter = Arc::new(AtomicUsize::new(0));
210        let config = LoopConfig::new(LoopMode::Count(3));
211        let mut service = LoopService::new(config, counter_pipeline(Arc::clone(&counter)));
212
213        let exchange = Exchange::new(Message::new("test"));
214        let result = service.ready().await.unwrap().call(exchange).await;
215
216        assert!(result.is_ok());
217        assert_eq!(counter.load(Ordering::SeqCst), 3);
218    }
219
220    #[tokio::test]
221    async fn test_loop_count_sets_properties() {
222        let seen_indices = Arc::new(Mutex::new(Vec::<u64>::new()));
223        let seen_indices_for_pipeline = Arc::clone(&seen_indices);
224
225        let pipeline = BoxProcessor::from_fn(move |exchange: Exchange| {
226            let seen_indices = Arc::clone(&seen_indices_for_pipeline);
227            Box::pin(async move {
228                if let Some(index) = exchange.property(CAMEL_LOOP_INDEX).and_then(|v| v.as_u64()) {
229                    seen_indices.lock().unwrap().push(index);
230                }
231                Ok(exchange)
232            })
233        });
234
235        let config = LoopConfig::new(LoopMode::Count(3));
236        let mut service = LoopService::new(config, pipeline);
237
238        let exchange = Exchange::new(Message::new("test"));
239        let result = service.ready().await.unwrap().call(exchange).await.unwrap();
240
241        assert_eq!(*seen_indices.lock().unwrap(), vec![0, 1, 2]);
242        assert_eq!(
243            result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()),
244            Some(3)
245        );
246    }
247
248    #[tokio::test]
249    async fn test_loop_count_zero_is_noop() {
250        let config = LoopConfig::new(LoopMode::Count(0));
251        let mut service = LoopService::new(config, identity_pipeline());
252
253        let exchange = Exchange::new(Message::new("test"));
254        let result = service.ready().await.unwrap().call(exchange).await.unwrap();
255
256        assert_eq!(result.input.body.as_text(), Some("test"));
257        assert_eq!(
258            result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()),
259            Some(0)
260        );
261        assert!(result.property(CAMEL_LOOP_INDEX).is_none());
262    }
263
264    #[tokio::test]
265    async fn test_loop_while_stops_when_predicate_false() {
266        let counter = Arc::new(AtomicUsize::new(0));
267
268        let predicate = Arc::new(|exchange: &Exchange| {
269            exchange
270                .property("iterations")
271                .and_then(|v| v.as_u64())
272                .unwrap_or(0)
273                < 2
274        });
275
276        let counter_for_pipeline = Arc::clone(&counter);
277        let pipeline = BoxProcessor::from_fn(move |mut exchange: Exchange| {
278            let counter = Arc::clone(&counter_for_pipeline);
279            Box::pin(async move {
280                let current = exchange
281                    .property("iterations")
282                    .and_then(|v| v.as_u64())
283                    .unwrap_or(0);
284                exchange.set_property("iterations", current + 1);
285                counter.fetch_add(1, Ordering::SeqCst);
286                Ok(exchange)
287            })
288        });
289
290        let config = LoopConfig::new(LoopMode::While(predicate));
291        let mut service = LoopService::new(config, pipeline);
292
293        let exchange = Exchange::new(Message::new("test"));
294        let result = service.ready().await.unwrap().call(exchange).await.unwrap();
295
296        assert_eq!(counter.load(Ordering::SeqCst), 2);
297        assert_eq!(
298            result.property("iterations").and_then(|v| v.as_u64()),
299            Some(2)
300        );
301        assert_eq!(
302            result.property(CAMEL_LOOP_INDEX).and_then(|v| v.as_u64()),
303            Some(1)
304        );
305        assert_eq!(
306            result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()),
307            Some(0)
308        );
309    }
310
311    #[tokio::test]
312    async fn test_loop_while_respects_max_iterations() {
313        let counter = Arc::new(AtomicUsize::new(0));
314        let predicate = Arc::new(|_exchange: &Exchange| true);
315        let config = LoopConfig::new(LoopMode::While(predicate));
316        let mut service = LoopService::new(config, counter_pipeline(Arc::clone(&counter)));
317
318        let exchange = Exchange::new(Message::new("test"));
319        let result = service.ready().await.unwrap().call(exchange).await;
320
321        assert!(result.is_ok());
322        assert_eq!(counter.load(Ordering::SeqCst), MAX_LOOP_ITERATIONS);
323    }
324
325    #[tokio::test]
326    async fn test_loop_error_propagation() {
327        let pipeline = BoxProcessor::from_fn(|_exchange: Exchange| {
328            Box::pin(async { Err(CamelError::ProcessorError("boom".into())) })
329        });
330
331        let config = LoopConfig::new(LoopMode::Count(3));
332        let mut service = LoopService::new(config, pipeline);
333
334        let exchange = Exchange::new(Message::new("test"));
335        let result = service.ready().await.unwrap().call(exchange).await;
336
337        assert!(matches!(result, Err(CamelError::ProcessorError(msg)) if msg == "boom"));
338    }
339
340    // ── D-M9 Batch 1: LoopMode::Count(n) clamped to MAX_LOOP_ITERATIONS ──
341
342    /// D-M9: a `Count(u32::MAX as usize)` (or any value above
343    /// `MAX_LOOP_ITERATIONS`) is clamped to `MAX_LOOP_ITERATIONS` and
344    /// runs at most that many iterations. Without the clamp, a malicious
345    /// or typo'd `count: 4294967295` would exhaust CPU. The pre-existing
346    /// `Count(3)` test continues to pass — small values are unaffected.
347    #[tokio::test]
348    async fn test_loop_count_clamped_to_max_iterations() {
349        let counter = Arc::new(AtomicUsize::new(0));
350        let config = LoopConfig::new(LoopMode::Count(usize::MAX));
351        let mut service = LoopService::new(config, counter_pipeline(Arc::clone(&counter)));
352
353        let exchange = Exchange::new(Message::new("test"));
354        let result = service.ready().await.unwrap().call(exchange).await;
355
356        assert!(result.is_ok());
357        // Must run exactly MAX_LOOP_ITERATIONS, not u32::MAX iterations.
358        assert_eq!(counter.load(Ordering::SeqCst), MAX_LOOP_ITERATIONS);
359        // The CAMEL_LOOP_SIZE property is set to the *clamped* count.
360        assert_eq!(
361            result
362                .unwrap()
363                .property(CAMEL_LOOP_SIZE)
364                .and_then(|v| v.as_u64()),
365            Some(MAX_LOOP_ITERATIONS as u64)
366        );
367    }
368
369    #[tokio::test]
370    async fn test_loop_pipeline_chaining() {
371        let pipeline = BoxProcessor::from_fn(|mut exchange: Exchange| {
372            Box::pin(async move {
373                if let Body::Text(s) = &exchange.input.body {
374                    exchange.input.body = Body::Text(format!("{s}x"));
375                }
376                Ok(exchange)
377            })
378        });
379
380        let config = LoopConfig::new(LoopMode::Count(3));
381        let mut service = LoopService::new(config, pipeline);
382
383        let exchange = Exchange::new(Message::new("start"));
384        let result = service.ready().await.unwrap().call(exchange).await.unwrap();
385
386        assert_eq!(result.input.body.as_text(), Some("startxxx"));
387    }
388}
camel_processor/loop_eip.rs

camel_processor/
loop_eip.rs
