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

camel_processor/
loop_eip.rs
