camel_processor/
loop_eip.rs1use 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 exchange.set_property(CAMEL_LOOP_SIZE, Value::from(n as u64));
45 for i in 0..n {
46 exchange.set_property(CAMEL_LOOP_INDEX, Value::from(i as u64));
47 exchange = pipeline.ready().await?.call(exchange).await?;
48 }
49 }
50 LoopMode::While(ref predicate) => {
51 exchange.set_property(CAMEL_LOOP_SIZE, Value::from(0u64));
52 for i in 0..MAX_LOOP_ITERATIONS {
53 if !predicate(&exchange) {
54 break;
55 }
56 exchange.set_property(CAMEL_LOOP_INDEX, Value::from(i as u64));
57 exchange = pipeline.ready().await?.call(exchange).await?;
58 }
59 if predicate(&exchange) {
60 tracing::warn!(
61 "Loop while-mode hit MAX_LOOP_ITERATIONS ({}) safety guard. Predicate still true.",
62 MAX_LOOP_ITERATIONS
63 );
64 }
65 }
66 }
67 Ok(exchange)
68 })
69 }
70}
71
72#[cfg(test)]
73mod tests {
74 use std::sync::atomic::{AtomicUsize, Ordering};
75 use std::sync::{Arc, Mutex};
76
77 use camel_api::loop_eip::{LoopConfig, LoopMode, MAX_LOOP_ITERATIONS};
78 use camel_api::{
79 Body, BoxProcessor, BoxProcessorExt, CamelError, Exchange, IdentityProcessor, Message,
80 };
81 use tower::{Service, ServiceExt};
82
83 use super::{CAMEL_LOOP_INDEX, CAMEL_LOOP_SIZE, LoopService};
84
85 fn identity_pipeline() -> BoxProcessor {
86 BoxProcessor::new(IdentityProcessor)
87 }
88
89 fn counter_pipeline(counter: Arc<AtomicUsize>) -> BoxProcessor {
90 BoxProcessor::from_fn(move |exchange: Exchange| {
91 let counter = Arc::clone(&counter);
92 Box::pin(async move {
93 counter.fetch_add(1, Ordering::SeqCst);
94 Ok(exchange)
95 })
96 })
97 }
98
99 #[tokio::test]
100 async fn test_loop_count_iterates_n_times() {
101 let counter = Arc::new(AtomicUsize::new(0));
102 let config = LoopConfig::new(LoopMode::Count(3));
103 let mut service = LoopService::new(config, counter_pipeline(Arc::clone(&counter)));
104
105 let exchange = Exchange::new(Message::new("test"));
106 let result = service.ready().await.unwrap().call(exchange).await;
107
108 assert!(result.is_ok());
109 assert_eq!(counter.load(Ordering::SeqCst), 3);
110 }
111
112 #[tokio::test]
113 async fn test_loop_count_sets_properties() {
114 let seen_indices = Arc::new(Mutex::new(Vec::<u64>::new()));
115 let seen_indices_for_pipeline = Arc::clone(&seen_indices);
116
117 let pipeline = BoxProcessor::from_fn(move |exchange: Exchange| {
118 let seen_indices = Arc::clone(&seen_indices_for_pipeline);
119 Box::pin(async move {
120 if let Some(index) = exchange.property(CAMEL_LOOP_INDEX).and_then(|v| v.as_u64()) {
121 seen_indices.lock().unwrap().push(index);
122 }
123 Ok(exchange)
124 })
125 });
126
127 let config = LoopConfig::new(LoopMode::Count(3));
128 let mut service = LoopService::new(config, pipeline);
129
130 let exchange = Exchange::new(Message::new("test"));
131 let result = service.ready().await.unwrap().call(exchange).await.unwrap();
132
133 assert_eq!(*seen_indices.lock().unwrap(), vec![0, 1, 2]);
134 assert_eq!(result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()), Some(3));
135 }
136
137 #[tokio::test]
138 async fn test_loop_count_zero_is_noop() {
139 let config = LoopConfig::new(LoopMode::Count(0));
140 let mut service = LoopService::new(config, identity_pipeline());
141
142 let exchange = Exchange::new(Message::new("test"));
143 let result = service.ready().await.unwrap().call(exchange).await.unwrap();
144
145 assert_eq!(result.input.body.as_text(), Some("test"));
146 assert_eq!(result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()), Some(0));
147 assert!(result.property(CAMEL_LOOP_INDEX).is_none());
148 }
149
150 #[tokio::test]
151 async fn test_loop_while_stops_when_predicate_false() {
152 let counter = Arc::new(AtomicUsize::new(0));
153
154 let predicate = Arc::new(|exchange: &Exchange| {
155 exchange
156 .property("iterations")
157 .and_then(|v| v.as_u64())
158 .unwrap_or(0)
159 < 2
160 });
161
162 let counter_for_pipeline = Arc::clone(&counter);
163 let pipeline = BoxProcessor::from_fn(move |mut exchange: Exchange| {
164 let counter = Arc::clone(&counter_for_pipeline);
165 Box::pin(async move {
166 let current = exchange
167 .property("iterations")
168 .and_then(|v| v.as_u64())
169 .unwrap_or(0);
170 exchange.set_property("iterations", current + 1);
171 counter.fetch_add(1, Ordering::SeqCst);
172 Ok(exchange)
173 })
174 });
175
176 let config = LoopConfig::new(LoopMode::While(predicate));
177 let mut service = LoopService::new(config, pipeline);
178
179 let exchange = Exchange::new(Message::new("test"));
180 let result = service.ready().await.unwrap().call(exchange).await.unwrap();
181
182 assert_eq!(counter.load(Ordering::SeqCst), 2);
183 assert_eq!(result.property("iterations").and_then(|v| v.as_u64()), Some(2));
184 assert_eq!(result.property(CAMEL_LOOP_INDEX).and_then(|v| v.as_u64()), Some(1));
185 assert_eq!(result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()), Some(0));
186 }
187
188 #[tokio::test]
189 async fn test_loop_while_respects_max_iterations() {
190 let counter = Arc::new(AtomicUsize::new(0));
191 let predicate = Arc::new(|_exchange: &Exchange| true);
192 let config = LoopConfig::new(LoopMode::While(predicate));
193 let mut service = LoopService::new(config, counter_pipeline(Arc::clone(&counter)));
194
195 let exchange = Exchange::new(Message::new("test"));
196 let result = service.ready().await.unwrap().call(exchange).await;
197
198 assert!(result.is_ok());
199 assert_eq!(counter.load(Ordering::SeqCst), MAX_LOOP_ITERATIONS);
200 }
201
202 #[tokio::test]
203 async fn test_loop_error_propagation() {
204 let pipeline = BoxProcessor::from_fn(|_exchange: Exchange| {
205 Box::pin(async { Err(CamelError::ProcessorError("boom".into())) })
206 });
207
208 let config = LoopConfig::new(LoopMode::Count(3));
209 let mut service = LoopService::new(config, pipeline);
210
211 let exchange = Exchange::new(Message::new("test"));
212 let result = service.ready().await.unwrap().call(exchange).await;
213
214 assert!(matches!(result, Err(CamelError::ProcessorError(msg)) if msg == "boom"));
215 }
216
217 #[tokio::test]
218 async fn test_loop_pipeline_chaining() {
219 let pipeline = BoxProcessor::from_fn(|mut exchange: Exchange| {
220 Box::pin(async move {
221 if let Body::Text(s) = &exchange.input.body {
222 exchange.input.body = Body::Text(format!("{s}x"));
223 }
224 Ok(exchange)
225 })
226 });
227
228 let config = LoopConfig::new(LoopMode::Count(3));
229 let mut service = LoopService::new(config, pipeline);
230
231 let exchange = Exchange::new(Message::new("start"));
232 let result = service.ready().await.unwrap().call(exchange).await.unwrap();
233
234 assert_eq!(result.input.body.as_text(), Some("startxxx"));
235 }
236
237}