camel_core/lifecycle/adapters/
exchange_uow.rs

1// crates/camel-core/src/lifecycle/adapters/exchange_uow.rs
2//! Exchange Unit of Work Tower layer.
3
4use std::future::Future;
5use std::pin::Pin;
6use std::sync::Arc;
7use std::task::{Context, Poll};
8use std::time::Duration;
9
10use tower::{Layer, Service, ServiceExt};
11
12use camel_api::{BoxProcessor, CamelError, Exchange, InFlightClaim, InFlightGauge};
13
14// ─── Layer ───────────────────────────────────────────────────────────────────
15
16#[derive(Clone)]
17pub struct ExchangeUoWLayer {
18    counter: Arc<InFlightGauge>,
19    on_complete_producer: Option<BoxProcessor>,
20    on_failure_producer: Option<BoxProcessor>,
21}
22
23impl ExchangeUoWLayer {
24    pub fn new(
25        counter: Arc<InFlightGauge>,
26        on_complete_producer: Option<BoxProcessor>,
27        on_failure_producer: Option<BoxProcessor>,
28    ) -> Self {
29        Self {
30            counter,
31            on_complete_producer,
32            on_failure_producer,
33        }
34    }
35}
36
37impl<S> Layer<S> for ExchangeUoWLayer
38where
39    S: Service<Exchange, Response = Exchange, Error = CamelError> + Send + Clone + 'static,
40    S::Future: Send + 'static,
41{
42    type Service = ExchangeUoW<S>;
43
44    fn layer(&self, inner: S) -> Self::Service {
45        ExchangeUoW {
46            inner,
47            counter: Arc::clone(&self.counter),
48            on_complete_producer: self.on_complete_producer.clone(),
49            on_failure_producer: self.on_failure_producer.clone(),
50        }
51    }
52}
53
54// ─── Service ─────────────────────────────────────────────────────────────────
55
56#[derive(Clone)]
57pub struct ExchangeUoW<S> {
58    inner: S,
59    counter: Arc<InFlightGauge>,
60    on_complete_producer: Option<BoxProcessor>,
61    on_failure_producer: Option<BoxProcessor>,
62}
63
64impl<S> Service<Exchange> for ExchangeUoW<S>
65where
66    S: Service<Exchange, Response = Exchange, Error = CamelError> + Send + Clone + 'static,
67    S::Future: Send + 'static,
68{
69    type Response = Exchange;
70    type Error = CamelError;
71    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
72
73    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
74        self.inner.poll_ready(cx)
75    }
76
77    fn call(&mut self, exchange: Exchange) -> Self::Future {
78        let mut inner = self.inner.clone();
79        let counter = Arc::clone(&self.counter);
80        let on_complete = self.on_complete_producer.clone();
81        let on_failure = self.on_failure_producer.clone();
82
83        Box::pin(async move {
84            // The claim increments at attach and decrements exactly once
85            // when the future completes (RAII) — same lifetime as the
86            // former fetch_add/guard pair.
87            let _claim = InFlightClaim::attach(&counter);
88            let original = exchange.clone();
89
90            let result = match inner.ready().await {
91                Ok(svc) => svc.call(exchange).await,
92                Err(err) => {
93                    // Infrastructure failure: inner service not ready.
94                    // Pass `original` (pre-call clone) since no post-processing occurred.
95                    fire_hook(on_failure.clone(), Some(original)).await;
96                    return Err(err);
97                }
98            };
99
100            match &result {
101                Err(_) => {
102                    // Inner call returned Err: pass `original` (pre-call clone), no output exchange.
103                    fire_hook(on_failure, Some(original)).await;
104                }
105                Ok(ex) if ex.has_error() => {
106                    // Exchange-level error: pass the post-processing exchange so the hook
107                    // can inspect the error flag and any headers/body set by the pipeline.
108                    fire_hook(on_failure, Some(ex.clone())).await;
109                }
110                Ok(ex) => {
111                    fire_hook(on_complete, Some(ex.clone())).await;
112                }
113            }
114
115            result
116        })
117    }
118}
119
120async fn fire_hook(producer: Option<BoxProcessor>, exchange: Option<Exchange>) {
121    let (Some(mut producer), Some(ex)) = (producer, exchange) else {
122        return;
123    };
124    let fire = async move {
125        if let Err(e) = producer.ready().await {
126            tracing::warn!(error = %e, "UoW hook producer not ready");
127            return;
128        }
129        if let Err(e) = producer.call(ex).await {
130            tracing::warn!(error = %e, "UoW hook producer call failed");
131        }
132    };
133    if tokio::time::timeout(Duration::from_secs(1), fire)
134        .await
135        .is_err()
136    {
137        tracing::warn!("UoW hook timed out after 1s (error swallowed)");
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use camel_api::{BoxProcessorExt, Message};
145    use std::sync::atomic::{AtomicU64, Ordering};
146
147    fn make_exchange() -> Exchange {
148        Exchange::new(Message::new("test"))
149    }
150
151    fn identity() -> BoxProcessor {
152        BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }))
153    }
154
155    fn failing() -> BoxProcessor {
156        BoxProcessor::from_fn(|_| {
157            Box::pin(async { Err(CamelError::ProcessorError("boom".into())) })
158        })
159    }
160
161    fn error_exchange_proc() -> BoxProcessor {
162        BoxProcessor::from_fn(|mut ex: Exchange| {
163            Box::pin(async move {
164                ex.set_error(CamelError::ProcessorError("exchange error".into()));
165                Ok(ex)
166            })
167        })
168    }
169
170    #[tokio::test]
171    async fn counter_increments_then_decrements_on_success() {
172        let counter = Arc::new(InFlightGauge::new());
173        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), None, None);
174        let svc = layer.layer(identity());
175        assert_eq!(counter.total(), 0);
176        let _ = tower::ServiceExt::oneshot(svc, make_exchange())
177            .await
178            .unwrap();
179        assert_eq!(counter.total(), 0);
180    }
181
182    #[tokio::test]
183    async fn counter_decrements_on_inner_error() {
184        let counter = Arc::new(InFlightGauge::new());
185        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), None, None);
186        let svc = layer.layer(failing());
187        let result = tower::ServiceExt::oneshot(svc, make_exchange()).await;
188        assert!(result.is_err());
189        assert_eq!(counter.total(), 0);
190    }
191
192    #[tokio::test]
193    async fn on_complete_fires_on_success() {
194        let fired = Arc::new(AtomicU64::new(0));
195        let fired_clone = Arc::clone(&fired);
196        let hook = BoxProcessor::from_fn(move |ex| {
197            fired_clone.fetch_add(1, Ordering::Relaxed);
198            Box::pin(async move { Ok(ex) })
199        });
200        let counter = Arc::new(InFlightGauge::new());
201        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), Some(hook), None);
202        let _ = tower::ServiceExt::oneshot(layer.layer(identity()), make_exchange())
203            .await
204            .unwrap();
205        assert_eq!(fired.load(Ordering::Relaxed), 1);
206    }
207
208    #[tokio::test]
209    async fn on_failure_fires_on_inner_error() {
210        let fired = Arc::new(AtomicU64::new(0));
211        let fired_clone = Arc::clone(&fired);
212        let hook = BoxProcessor::from_fn(move |ex| {
213            fired_clone.fetch_add(1, Ordering::Relaxed);
214            Box::pin(async move { Ok(ex) })
215        });
216        let counter = Arc::new(InFlightGauge::new());
217        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), None, Some(hook));
218        let _ = tower::ServiceExt::oneshot(layer.layer(failing()), make_exchange()).await;
219        assert_eq!(fired.load(Ordering::Relaxed), 1);
220    }
221
222    #[tokio::test]
223    async fn on_failure_fires_on_exchange_error() {
224        let fired = Arc::new(AtomicU64::new(0));
225        let fired_clone = Arc::clone(&fired);
226        let hook = BoxProcessor::from_fn(move |ex| {
227            fired_clone.fetch_add(1, Ordering::Relaxed);
228            Box::pin(async move { Ok(ex) })
229        });
230        let counter = Arc::new(InFlightGauge::new());
231        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), None, Some(hook));
232        let _ = tower::ServiceExt::oneshot(layer.layer(error_exchange_proc()), make_exchange())
233            .await
234            .unwrap();
235        assert_eq!(fired.load(Ordering::Relaxed), 1);
236    }
237
238    #[tokio::test]
239    async fn on_failure_fires_when_poll_ready_fails() {
240        use std::future::Future;
241        use std::pin::Pin;
242        use std::sync::Arc;
243        use std::sync::atomic::AtomicU64;
244        use std::task::{Context, Poll};
245        use tower::Service;
246
247        #[derive(Clone)]
248        struct FailReadySvc {
249            polls: Arc<AtomicU64>,
250        }
251        impl Service<Exchange> for FailReadySvc {
252            type Response = Exchange;
253            type Error = CamelError;
254            type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
255            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), CamelError>> {
256                let n = self.polls.fetch_add(1, Ordering::Relaxed);
257                if n == 0 {
258                    Poll::Ready(Ok(()))
259                } else {
260                    Poll::Ready(Err(CamelError::ProcessorError("not ready".into())))
261                }
262            }
263            fn call(&mut self, ex: Exchange) -> Self::Future {
264                Box::pin(async move { Ok(ex) })
265            }
266        }
267
268        let fired = Arc::new(AtomicU64::new(0));
269        let fired_clone = Arc::clone(&fired);
270        let hook = BoxProcessor::from_fn(move |ex| {
271            fired_clone.fetch_add(1, Ordering::Relaxed);
272            Box::pin(async move { Ok(ex) })
273        });
274        let counter = Arc::new(InFlightGauge::new());
275        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), None, Some(hook));
276        let svc = layer.layer(FailReadySvc {
277            polls: Arc::new(AtomicU64::new(0)),
278        });
279        let result = tower::ServiceExt::oneshot(svc, make_exchange()).await;
280        assert!(result.is_err());
281        assert_eq!(
282            fired.load(Ordering::Relaxed),
283            1,
284            "on_failure must fire when poll_ready fails"
285        );
286        assert_eq!(counter.total(), 0);
287    }
288
289    #[tokio::test]
290    async fn on_complete_does_not_fire_on_exchange_error() {
291        let fired = Arc::new(AtomicU64::new(0));
292        let fired_clone = Arc::clone(&fired);
293        let hook = BoxProcessor::from_fn(move |ex| {
294            fired_clone.fetch_add(1, Ordering::Relaxed);
295            Box::pin(async move { Ok(ex) })
296        });
297        let counter = Arc::new(InFlightGauge::new());
298        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), Some(hook), None);
299        let _ = tower::ServiceExt::oneshot(layer.layer(error_exchange_proc()), make_exchange())
300            .await
301            .unwrap();
302        assert_eq!(fired.load(Ordering::Relaxed), 0);
303    }
304
305    #[tokio::test]
306    async fn hook_error_does_not_fail_exchange() {
307        let bad_hook = BoxProcessor::from_fn(|_| {
308            Box::pin(async { Err(CamelError::ProcessorError("hook failed".into())) })
309        });
310        let counter = Arc::new(InFlightGauge::new());
311        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), Some(bad_hook), None);
312        let result = tower::ServiceExt::oneshot(layer.layer(identity()), make_exchange()).await;
313        assert!(
314            result.is_ok(),
315            "hook error must not fail exchange: {:?}",
316            result
317        );
318    }
319
320    #[tokio::test]
321    async fn on_complete_fires_on_stopped() {
322        // ADR-0024: Stop arrives as Ok(ex) at the Tower boundary, so ExchangeUoW's
323        // `Ok(ex) => fire_hook(on_complete, ...)` arm fires for Stop.
324        use crate::lifecycle::adapters::route_compiler::{
325            PipelineRuntimeCtx, compose_pipeline_with_handler,
326        };
327        use crate::lifecycle::adapters::step_compilers::CompiledStep;
328
329        let fired = Arc::new(AtomicU64::new(0));
330        let fired_clone = Arc::clone(&fired);
331        let hook = BoxProcessor::from_fn(move |ex| {
332            fired_clone.fetch_add(1, Ordering::Relaxed);
333            Box::pin(async move { Ok(ex) })
334        });
335        let counter = Arc::new(InFlightGauge::new());
336        let layer = ExchangeUoWLayer::new(Arc::clone(&counter), Some(hook), None);
337
338        // Top-level pipeline with Stop — maps Stop to Ok(ex).
339        let stop_pipeline: BoxProcessor = compose_pipeline_with_handler(
340            vec![CompiledStep::Stop],
341            None,
342            PipelineRuntimeCtx::compile_time(),
343        );
344
345        let svc = layer.layer(stop_pipeline);
346        let ex = Exchange::new(Message::new("payload"));
347        let result = tower::ServiceExt::oneshot(svc, ex).await;
348
349        assert!(result.is_ok(), "Stop must arrive as Ok at the UoW layer");
350        assert_eq!(
351            fired.load(Ordering::Relaxed),
352            1,
353            "on_complete MUST fire for Stop (same as Completed)"
354        );
355    }
356}
camel_core/lifecycle/adapters/exchange_uow.rs

camel_core/lifecycle/adapters/
exchange_uow.rs
