camel_processor/
do_try.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 camel_api::error_handler::ExceptionDisposition;
6use camel_api::exchange::PROPERTY_EXCEPTION_HANDLED;
7use camel_api::{BoxProcessor, CamelError, Exchange, FilterPredicate};
8use tower::Service;
9use tower::ServiceExt;
10
11/// Matcher for a `doCatch` clause.
12#[derive(Clone)]
13pub enum CatchMatcher {
14    /// Match by CamelError variant name (e.g. ["ProcessorError", "Io"]).
15    /// `"*"` matches any variant — equivalent to Camel's `doCatch(Throwable.class)`.
16    ByVariant(Vec<String>),
17    /// Match by predicate over the Exchange.
18    Predicate(FilterPredicate),
19}
20
21impl CatchMatcher {
22    /// Returns `true` if this matcher matches the given error and exchange.
23    pub fn matches(&self, err: &CamelError, ex: &Exchange) -> bool {
24        match self {
25            CatchMatcher::ByVariant(names) => {
26                if names.iter().any(|n| n == "*") {
27                    return true;
28                }
29                names.iter().any(|n| n == err.variant_name())
30            }
31            CatchMatcher::Predicate(p) => p(ex),
32        }
33    }
34}
35
36/// A single `doCatch` clause.
37#[derive(Clone)]
38pub struct CatchClause {
39    /// The main matcher (variant-name list or predicate).
40    pub matcher: CatchMatcher,
41    /// Optional sub-predicate evaluated AFTER the main matcher passes.
42    pub on_when: Option<FilterPredicate>,
43    /// Sub-pipeline executed when the clause matches.
44    pub steps: Vec<BoxProcessor>,
45    /// ADR-0019 disposition: Handled (default), Propagate, or Continued (rejected at parse time).
46    /// In YAML, use lowercase: `handled`, `propagate`, `continued`.
47    pub disposition: ExceptionDisposition,
48}
49
50/// The `doTry` processor. Wrap with `BoxProcessor::new(DoTryService::new(...))`.
51#[derive(Clone)]
52pub struct DoTryService {
53    /// Steps in the try block.
54    pub try_steps: Vec<BoxProcessor>,
55    /// Catch clauses evaluated first-match-wins.
56    pub catch_clauses: Vec<CatchClause>,
57    /// Steps in the finally block (empty = no finally).
58    pub finally_steps: Vec<BoxProcessor>,
59    /// Optional onWhen predicate for finally.
60    pub finally_on_when: Option<FilterPredicate>,
61}
62
63impl DoTryService {
64    /// Create a new `DoTryService` with the given try steps.
65    pub fn new(try_steps: Vec<BoxProcessor>) -> Self {
66        Self {
67            try_steps,
68            catch_clauses: Vec::new(),
69            finally_steps: Vec::new(),
70            finally_on_when: None,
71        }
72    }
73
74    /// Full constructor used by the compile pipeline (Task 10b control_flow.rs).
75    /// Builder API (Task 8) constructs via `new()` + field mutation.
76    pub fn with_catch_and_finally(
77        try_steps: Vec<BoxProcessor>,
78        catch_clauses: Vec<CatchClause>,
79        finally_steps: Vec<BoxProcessor>,
80        finally_on_when: Option<FilterPredicate>,
81    ) -> Self {
82        Self {
83            try_steps,
84            catch_clauses,
85            finally_steps,
86            finally_on_when,
87        }
88    }
89}
90
91/// Run a sequence of steps, preserving the last exchange state on error.
92/// Returns `Err((last_ex, err))` so DoTry can populate exception properties.
93async fn run_pipeline(
94    steps: Vec<BoxProcessor>,
95    mut ex: Exchange,
96) -> Result<Exchange, (Exchange, CamelError)> {
97    for mut svc in steps {
98        match svc.ready().await {
99            Ok(ready) => {
100                let snapshot = ex.clone();
101                match ready.call(ex).await {
102                    Ok(new_ex) => ex = new_ex,
103                    Err(err) => return Err((snapshot, err)),
104                }
105            }
106            Err(err) => return Err((ex, err)),
107        }
108    }
109    Ok(ex)
110}
111
112/// Run the finally block. Camel parity for finally-throws:
113/// - If finally succeeds: return its exchange.
114/// - If finally throws AND there was a previous error: restore previous (log finally_err).
115/// - If finally throws AND no previous error: propagate finally_err.
116async fn run_finally(
117    finally_steps: Vec<BoxProcessor>,
118    finally_on_when: Option<FilterPredicate>,
119    ex: Exchange,
120    previous_err: Option<CamelError>,
121) -> Result<Exchange, CamelError> {
122    if finally_steps.is_empty() {
123        return Ok(ex);
124    }
125    if let Some(on_when) = &finally_on_when
126        && !on_when(&ex)
127    {
128        return Ok(ex);
129    }
130    match run_pipeline(finally_steps, ex).await {
131        Ok(ex) => Ok(ex),
132        Err((_, finally_err)) => match previous_err {
133            Some(prev) => {
134                tracing::warn!(
135                    finally_error = %finally_err,
136                    previous_error = %prev,
137                    "doFinally threw; restoring previous exception (Camel parity)"
138                );
139                Err(prev)
140            }
141            None => {
142                tracing::warn!(error = %finally_err, "doFinally threw");
143                Err(finally_err)
144            }
145        },
146    }
147}
148
149impl tower::Service<Exchange> for DoTryService {
150    type Response = Exchange;
151    type Error = CamelError;
152    type Future = std::pin::Pin<
153        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
154    >;
155
156    fn poll_ready(
157        &mut self,
158        _cx: &mut std::task::Context<'_>,
159    ) -> std::task::Poll<Result<(), Self::Error>> {
160        std::task::Poll::Ready(Ok(()))
161    }
162
163    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
164        // Clear stale CamelExceptionHandled marker from prior handlers in the same route.
165        // Exchange::clear_error() does NOT touch HANDLED, so direct map access is required.
166        exchange.properties.remove(PROPERTY_EXCEPTION_HANDLED);
167
168        let try_steps = self.try_steps.clone();
169        let catch_clauses = self.catch_clauses.clone();
170        let finally_steps = self.finally_steps.clone();
171        let finally_on_when = self.finally_on_when.clone();
172
173        Box::pin(async move {
174            let try_result = run_pipeline(try_steps, exchange).await;
175            match try_result {
176                Ok(ex) => run_finally(finally_steps, finally_on_when, ex, None).await,
177                Err((failed_ex, original_err)) => {
178                    let mut ex = failed_ex;
179                    ex.set_error(original_err.clone());
180
181                    for clause in catch_clauses {
182                        let CatchClause {
183                            matcher,
184                            on_when,
185                            steps,
186                            disposition,
187                        } = clause;
188                        if !matcher.matches(&original_err, &ex) {
189                            continue;
190                        }
191                        if let Some(ref on_when) = on_when
192                            && !on_when(&ex)
193                        {
194                            continue;
195                        }
196
197                        let catch_result = run_pipeline(steps, ex.clone()).await;
198
199                        return match catch_result {
200                            Ok(ok_ex) => {
201                                // Determine previous-error threading based on disposition.
202                                // IMPORTANT: do NOT call handle_error() before run_finally() —
203                                // handle_error() calls clear_error() which removes
204                                // PROPERTY_EXCEPTION_MESSAGE/KIND/CAUGHT, preventing finally
205                                // steps from inspecting the caught exception.
206                                //
207                                // disposition semantics (ADR-0019 strict):
208                                //   Handled    -> catch output is final, no propagation
209                                //   Propagate  -> catch ran for side-effects, original propagates
210                                //   Continued  -> rejected at parse time (defensive: treat as
211                                //                 Propagate + log if we ever reach runtime)
212                                let prev = match disposition {
213                                    ExceptionDisposition::Handled => None,
214                                    ExceptionDisposition::Propagate => Some(original_err.clone()),
215                                    ExceptionDisposition::Continued => {
216                                        tracing::warn!(
217                                            "ExceptionDisposition::Continued reached doTry runtime; \
218                                             treating as Propagate. Should have been rejected at parse time."
219                                        );
220                                        Some(original_err.clone())
221                                    }
222                                };
223                                let mut ex = run_finally(
224                                    finally_steps.clone(),
225                                    finally_on_when.clone(),
226                                    ok_ex,
227                                    prev,
228                                )
229                                .await?;
230                                // AFTER finally has run (and had access to exception props),
231                                // apply handle_error() for Handled disposition to clear the
232                                // error state and set CamelExceptionHandled=true marker.
233                                if matches!(disposition, ExceptionDisposition::Handled) {
234                                    ex.handle_error();
235                                }
236                                match disposition {
237                                    ExceptionDisposition::Handled => Ok(ex),
238                                    _ => Err(original_err),
239                                }
240                            }
241                            Err((catch_ex, catch_err)) => {
242                                // Catch threw. Run finally with previous=catch_err.
243                                // Per Camel parity, if finally itself throws, catch_err is restored.
244                                let _ex = run_finally(
245                                    finally_steps.clone(),
246                                    finally_on_when.clone(),
247                                    catch_ex,
248                                    Some(catch_err.clone()),
249                                )
250                                .await?;
251                                Err(catch_err)
252                            }
253                        };
254                    }
255
256                    // No catch matched. Run finally with previous=original. Propagate original.
257                    let _ex = run_finally(
258                        finally_steps,
259                        finally_on_when,
260                        ex,
261                        Some(original_err.clone()),
262                    )
263                    .await?;
264                    Err(original_err)
265                }
266            }
267        })
268    }
269}
270
271// ── DoTrySegment (ADR-0025 OutcomePipeline) ──────────────────────────────
272
273/// Compilable segment for a `doCatch` clause within a `DoTrySegment`.
274///
275/// `disposition` controls outcome when the catch body completes:
276#[cfg(test)]
277mod tests {
278    use super::*;
279    use camel_api::{BoxProcessor, BoxProcessorExt};
280    use std::sync::Arc;
281    use std::sync::atomic::{AtomicU32, Ordering};
282
283    fn passthrough() -> BoxProcessor {
284        BoxProcessor::from_fn(move |ex| Box::pin(async move { Ok(ex) }))
285    }
286
287    fn record_call(flag: Arc<AtomicU32>) -> BoxProcessor {
288        BoxProcessor::from_fn(move |ex| {
289            let f = flag.clone();
290            Box::pin(async move {
291                f.fetch_add(1, Ordering::SeqCst);
292                Ok(ex)
293            })
294        })
295    }
296
297    fn always_fail(err: CamelError) -> BoxProcessor {
298        BoxProcessor::from_fn(move |_ex| {
299            let e = err.clone();
300            Box::pin(async move { Err(e) })
301        })
302    }
303
304    #[tokio::test]
305    async fn happy_path_try_succeeds_finally_runs() {
306        let finally_flag = Arc::new(AtomicU32::new(0));
307        let mut svc = DoTryService::new(vec![passthrough()]);
308        svc.finally_steps = vec![record_call(finally_flag.clone())];
309
310        let mut boxed = BoxProcessor::new(svc);
311        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
312        assert!(result.is_ok());
313        assert_eq!(finally_flag.load(Ordering::SeqCst), 1);
314    }
315
316    #[tokio::test]
317    async fn catch_by_variant_handled_returns_ok() {
318        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
319        let mut svc = DoTryService::new(vec![try_step]);
320        svc.catch_clauses.push(CatchClause {
321            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
322            on_when: None,
323            steps: vec![passthrough()],
324            disposition: ExceptionDisposition::Handled,
325        });
326
327        let mut boxed = BoxProcessor::new(svc);
328        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
329        assert!(result.is_ok(), "Handled must return Ok");
330        let ex = result.unwrap();
331        assert_eq!(
332            ex.properties.get(PROPERTY_EXCEPTION_HANDLED),
333            Some(&camel_api::Value::Bool(true)),
334            "CamelExceptionHandled must be set via handle_error()"
335        );
336    }
337
338    #[tokio::test]
339    async fn catch_by_variant_propagate_runs_side_effects_and_rethrows() {
340        let original = CamelError::ProcessorError("boom".into());
341        let try_step = always_fail(original.clone());
342        let side_effect = Arc::new(AtomicU32::new(0));
343        let catch_step = record_call(side_effect.clone());
344        let mut svc = DoTryService::new(vec![try_step]);
345        svc.catch_clauses.push(CatchClause {
346            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
347            on_when: None,
348            steps: vec![catch_step],
349            disposition: ExceptionDisposition::Propagate,
350        });
351
352        let mut boxed = BoxProcessor::new(svc);
353        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
354        assert!(result.is_err(), "Propagate must rethrow original");
355        assert!(matches!(result.unwrap_err(), CamelError::ProcessorError(_)));
356        assert_eq!(
357            side_effect.load(Ordering::SeqCst),
358            1,
359            "catch branch must have run for side-effects"
360        );
361    }
362
363    #[tokio::test]
364    async fn catch_by_predicate_matches_via_exception_kind() {
365        let try_step = always_fail(CamelError::Io("disk full".into()));
366        let predicate: FilterPredicate = Arc::new(|ex: &Exchange| {
367            ex.properties
368                .get(camel_api::exchange::PROPERTY_EXCEPTION_KIND)
369                .map(|v| matches!(v, camel_api::Value::String(s) if s == "io"))
370                .unwrap_or(false)
371        });
372        let mut svc = DoTryService::new(vec![try_step]);
373        svc.catch_clauses.push(CatchClause {
374            matcher: CatchMatcher::Predicate(predicate),
375            on_when: None,
376            steps: vec![passthrough()],
377            disposition: ExceptionDisposition::Handled,
378        });
379
380        let mut boxed = BoxProcessor::new(svc);
381        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
382        assert!(
383            result.is_ok(),
384            "Predicate matcher must catch the error and Handled must return Ok"
385        );
386    }
387
388    #[tokio::test]
389    async fn on_when_filters_clause_and_next_evaluated() {
390        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
391        let first_call = Arc::new(AtomicU32::new(0));
392        let second_call = Arc::new(AtomicU32::new(0));
393
394        let mut svc = DoTryService::new(vec![try_step]);
395        svc.catch_clauses.push(CatchClause {
396            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
397            on_when: Some(Arc::new(|_ex| false)),
398            steps: vec![record_call(first_call.clone())],
399            disposition: ExceptionDisposition::Handled,
400        });
401        svc.catch_clauses.push(CatchClause {
402            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
403            on_when: None,
404            steps: vec![record_call(second_call.clone())],
405            disposition: ExceptionDisposition::Handled,
406        });
407
408        let mut boxed = BoxProcessor::new(svc);
409        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
410        assert_eq!(first_call.load(Ordering::SeqCst), 0);
411        assert_eq!(second_call.load(Ordering::SeqCst), 1);
412    }
413
414    #[tokio::test]
415    async fn first_match_wins_subsequent_clauses_not_evaluated() {
416        let try_step = always_fail(CamelError::Io("err".into()));
417        let first_call = Arc::new(AtomicU32::new(0));
418        let second_call = Arc::new(AtomicU32::new(0));
419
420        let mut svc = DoTryService::new(vec![try_step]);
421        svc.catch_clauses.push(CatchClause {
422            matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
423            on_when: None,
424            steps: vec![record_call(first_call.clone())],
425            disposition: ExceptionDisposition::Handled,
426        });
427        svc.catch_clauses.push(CatchClause {
428            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
429            on_when: None,
430            steps: vec![record_call(second_call.clone())],
431            disposition: ExceptionDisposition::Handled,
432        });
433
434        let mut boxed = BoxProcessor::new(svc);
435        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
436        assert_eq!(first_call.load(Ordering::SeqCst), 1);
437        assert_eq!(second_call.load(Ordering::SeqCst), 0);
438    }
439
440    #[tokio::test]
441    async fn no_clause_matches_propagates_original() {
442        let try_step = always_fail(CamelError::CircuitOpen("cb".into()));
443        let mut svc = DoTryService::new(vec![try_step]);
444        svc.catch_clauses.push(CatchClause {
445            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
446            on_when: None,
447            steps: vec![passthrough()],
448            disposition: ExceptionDisposition::Handled,
449        });
450
451        let mut boxed = BoxProcessor::new(svc);
452        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
453        assert!(result.is_err());
454        assert!(matches!(result.unwrap_err(), CamelError::CircuitOpen(_)));
455    }
456
457    #[tokio::test]
458    async fn catch_branch_throws_new_error_wins() {
459        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
460        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
461        let mut svc = DoTryService::new(vec![try_step]);
462        svc.catch_clauses.push(CatchClause {
463            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
464            on_when: None,
465            steps: vec![catch_step],
466            disposition: ExceptionDisposition::Handled,
467        });
468
469        let mut boxed = BoxProcessor::new(svc);
470        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
471        assert!(result.is_err());
472        assert!(matches!(result.unwrap_err(), CamelError::Io(_)));
473    }
474
475    #[tokio::test]
476    async fn finally_throws_with_no_previous_error_propagates_finally_error() {
477        let finally_step = always_fail(CamelError::Config("fin".into()));
478        let mut svc = DoTryService::new(vec![passthrough()]);
479        svc.finally_steps = vec![finally_step];
480
481        let mut boxed = BoxProcessor::new(svc);
482        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
483        assert!(result.is_err());
484        assert!(matches!(result.unwrap_err(), CamelError::Config(_)));
485    }
486
487    #[tokio::test]
488    async fn finally_throws_with_previous_error_restores_previous() {
489        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
490        let finally_step = always_fail(CamelError::Config("fin".into()));
491        let mut svc = DoTryService::new(vec![try_step]);
492        svc.finally_steps = vec![finally_step];
493
494        let mut boxed = BoxProcessor::new(svc);
495        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
496        assert!(result.is_err());
497        assert!(
498            matches!(result.unwrap_err(), CamelError::ProcessorError(_)),
499            "previous error must be restored when finally throws (Camel parity)"
500        );
501    }
502
503    #[tokio::test]
504    async fn finally_on_when_false_skips_finally() {
505        let finally_call = Arc::new(AtomicU32::new(0));
506        let mut svc = DoTryService::new(vec![passthrough()]);
507        svc.finally_steps = vec![record_call(finally_call.clone())];
508        svc.finally_on_when = Some(Arc::new(|_ex| false));
509
510        let mut boxed = BoxProcessor::new(svc);
511        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
512        assert_eq!(finally_call.load(Ordering::SeqCst), 0);
513    }
514
515    #[tokio::test]
516    async fn stale_handled_marker_cleared_on_entry() {
517        let mut ex = Exchange::default();
518        ex.set_property(PROPERTY_EXCEPTION_HANDLED, camel_api::Value::Bool(true));
519        let svc = DoTryService::new(vec![passthrough()]);
520        let mut boxed = BoxProcessor::new(svc);
521        let result = boxed.ready().await.unwrap().call(ex).await;
522        let ex = result.unwrap();
523        assert!(
524            !ex.properties.contains_key(PROPERTY_EXCEPTION_HANDLED),
525            "stale CamelExceptionHandled must be cleared on entry"
526        );
527    }
528
529    #[tokio::test]
530    async fn nested_do_try_inner_catch_does_not_leak_to_outer() {
531        let inner = {
532            let try_step = always_fail(CamelError::Io("inner".into()));
533            let mut d = DoTryService::new(vec![try_step]);
534            d.catch_clauses.push(CatchClause {
535                matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
536                on_when: None,
537                steps: vec![passthrough()],
538                disposition: ExceptionDisposition::Handled,
539            });
540            BoxProcessor::new(d)
541        };
542        let mut outer = DoTryService::new(vec![inner]);
543        outer.catch_clauses.push(CatchClause {
544            matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
545            on_when: None,
546            steps: vec![passthrough()],
547            disposition: ExceptionDisposition::Handled,
548        });
549
550        let mut boxed = BoxProcessor::new(outer);
551        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
552        assert!(
553            result.is_ok(),
554            "outer must see Ok because inner handled its own error"
555        );
556    }
557
558    #[tokio::test]
559    async fn catch_all_only_fires_when_no_specific_clause_matches() {
560        let try_step = always_fail(CamelError::Io("err".into()));
561        let processor_call = Arc::new(AtomicU32::new(0));
562        let catch_all_call = Arc::new(AtomicU32::new(0));
563
564        let mut svc = DoTryService::new(vec![try_step]);
565        // First clause (specific) targets ProcessorError — won't match Io error.
566        svc.catch_clauses.push(CatchClause {
567            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
568            on_when: None,
569            steps: vec![record_call(processor_call.clone())],
570            disposition: ExceptionDisposition::Handled,
571        });
572        // Second clause is the catch-all — should fire.
573        svc.catch_clauses.push(CatchClause {
574            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
575            on_when: None,
576            steps: vec![record_call(catch_all_call.clone())],
577            disposition: ExceptionDisposition::Handled,
578        });
579
580        let mut boxed = BoxProcessor::new(svc);
581        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
582        assert_eq!(
583            processor_call.load(Ordering::SeqCst),
584            0,
585            "specific ProcessorError clause must not fire on Io error"
586        );
587        assert_eq!(
588            catch_all_call.load(Ordering::SeqCst),
589            1,
590            "catch-all clause must fire when no specific clause matches"
591        );
592    }
593
594    #[tokio::test]
595    async fn catch_throws_with_finally_runs_finally_and_propagates_catch_err() {
596        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
597        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
598        let finally_flag = Arc::new(AtomicU32::new(0));
599        let finally_step = record_call(finally_flag.clone());
600
601        let mut svc = DoTryService::new(vec![try_step]);
602        svc.catch_clauses.push(CatchClause {
603            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
604            on_when: None,
605            steps: vec![catch_step],
606            disposition: ExceptionDisposition::Handled,
607        });
608        svc.finally_steps = vec![finally_step];
609
610        let mut boxed = BoxProcessor::new(svc);
611        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
612
613        assert!(result.is_err());
614        assert!(
615            matches!(result.unwrap_err(), CamelError::Io(_)),
616            "catch_err must propagate (not original ProcessorError)"
617        );
618        assert_eq!(
619            finally_flag.load(Ordering::SeqCst),
620            1,
621            "doFinally must run even when catch throws"
622        );
623    }
624
625    #[tokio::test]
626    async fn catch_throws_and_finally_throws_restores_catch_err() {
627        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
628        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
629        let finally_step = always_fail(CamelError::Config("fin-fail".into()));
630
631        let mut svc = DoTryService::new(vec![try_step]);
632        svc.catch_clauses.push(CatchClause {
633            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
634            on_when: None,
635            steps: vec![catch_step],
636            disposition: ExceptionDisposition::Handled,
637        });
638        svc.finally_steps = vec![finally_step];
639
640        let mut boxed = BoxProcessor::new(svc);
641        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
642
643        assert!(result.is_err());
644        assert!(
645            matches!(result.unwrap_err(), CamelError::Io(_)),
646            "catch_err (Io) must be restored over finally_err (Config) per Camel parity"
647        );
648    }
649
650    #[tokio::test]
651    async fn finally_on_when_false_with_previous_error_still_propagates_original() {
652        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
653        let finally_flag = Arc::new(AtomicU32::new(0));
654        let finally_step = record_call(finally_flag.clone());
655
656        let mut svc = DoTryService::new(vec![try_step]);
657        // No catch clauses → original error stays.
658        svc.finally_steps = vec![finally_step];
659        svc.finally_on_when = Some(Arc::new(|_ex| false));
660
661        let mut boxed = BoxProcessor::new(svc);
662        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
663
664        assert!(result.is_err());
665        assert!(
666            matches!(result.unwrap_err(), CamelError::ProcessorError(_)),
667            "original error must propagate even when finally_on_when skips finally"
668        );
669        assert_eq!(
670            finally_flag.load(Ordering::SeqCst),
671            0,
672            "doFinally must NOT run when on_when returns false"
673        );
674    }
675}
camel_processor/do_try.rs

camel_processor/
do_try.rs
