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

camel_processor/
do_try.rs
