1use crate::do_try::CatchMatcher;
7use crate::do_try::chain_predicate_error;
8use crate::error_handler::record_span_error;
9use camel_api::error_handler::ExceptionDisposition;
10use camel_api::outcome_pipeline::OutcomePipeline;
11use camel_api::pipeline_outcome::PipelineOutcome;
12use camel_api::{CamelError, Exchange, PredicateSource};
13use std::future::Future;
14use std::pin::Pin;
15
16#[derive(Clone)]
20pub struct CatchClauseSegment {
21 pub matcher: CatchMatcher,
22 pub on_when: Option<PredicateSource>,
23 pub body: camel_api::OutcomeSegment,
24 pub disposition: ExceptionDisposition,
25}
26
27#[derive(Clone)]
29pub struct FinallyClauseSegment {
30 pub on_when: Option<PredicateSource>,
31 pub body: camel_api::OutcomeSegment,
32}
33
34pub struct DoTrySegment {
40 pub try_body: camel_api::OutcomeSegment,
41 pub catches: Vec<CatchClauseSegment>,
42 pub finally: Option<FinallyClauseSegment>,
43}
44
45impl Clone for DoTrySegment {
46 fn clone(&self) -> Self {
47 Self {
48 try_body: self.try_body.clone(),
49 catches: self.catches.clone(),
50 finally: self.finally.clone(),
51 }
52 }
53}
54
55enum FinallyOutcome {
60 Stopped(Box<Exchange>),
61 Failed(CamelError),
62 PredicateFailed(CamelError),
65}
66
67#[allow(clippy::result_large_err)]
74async fn run_finally_body(
75 finally: &mut Option<FinallyClauseSegment>,
76 ex: Exchange,
77) -> Result<Exchange, FinallyOutcome> {
78 let Some(f) = finally.as_mut() else {
79 return Ok(ex);
80 };
81 if let Some(on_when) = &f.on_when {
82 match on_when.matches(&ex).await {
83 Ok(true) => {}
84 Ok(false) => return Ok(ex),
85 Err(err) => return Err(FinallyOutcome::PredicateFailed(err)),
86 }
87 }
88 match f.body.run(ex).await {
89 PipelineOutcome::Completed(e) => Ok(e),
90 PipelineOutcome::Stopped(e) => Err(FinallyOutcome::Stopped(Box::new(e))),
91 PipelineOutcome::Failed(e) => Err(FinallyOutcome::Failed(e)),
92 }
93}
94
95async fn fail_after_finally(
104 finally: &mut Option<FinallyClauseSegment>,
105 ex: Exchange,
106 failure: CamelError,
107) -> PipelineOutcome {
108 match run_finally_body(finally, ex).await {
109 Ok(_) => PipelineOutcome::Failed(failure),
110 Err(FinallyOutcome::Stopped(e)) => PipelineOutcome::Stopped(*e),
111 Err(FinallyOutcome::Failed(_finally_err)) => {
112 tracing::warn!(
113 error = %failure,
114 "doFinally threw; restoring previous error (Camel parity)"
115 );
116 PipelineOutcome::Failed(failure)
117 }
118 Err(FinallyOutcome::PredicateFailed(pred_err)) => PipelineOutcome::Failed(pred_err),
119 }
120}
121
122impl OutcomePipeline for DoTrySegment {
123 fn clone_box(&self) -> Box<dyn OutcomePipeline> {
124 Box::new(self.clone())
125 }
126
127 fn run<'a>(
128 &'a mut self,
129 exchange: Exchange,
130 ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
131 Box::pin(async move {
132 let exchange_for_unmatched = exchange.branch_clone();
135
136 let try_outcome = self.try_body.run(exchange).await;
138 let returned_ex = match try_outcome {
139 PipelineOutcome::Stopped(ex) => return PipelineOutcome::Stopped(ex),
140 PipelineOutcome::Completed(ex) => ex,
141 PipelineOutcome::Failed(err) => {
142 let mut current_ex = exchange_for_unmatched;
148 current_ex.set_error(err.clone());
149 let mut catch_matched = false;
153 for catch in self.catches.iter_mut() {
154 let matched = match catch.matcher.matches(&err, ¤t_ex).await {
155 Ok(m) => m,
156 Err(pred_err) => {
157 let chained = chain_predicate_error(pred_err, err);
158 return fail_after_finally(&mut self.finally, current_ex, chained)
159 .await;
160 }
161 };
162 if !matched {
163 continue;
164 }
165 if let Some(on_when) = &catch.on_when {
166 match on_when.matches(¤t_ex).await {
167 Ok(true) => {}
168 Ok(false) => continue,
169 Err(pred_err) => {
170 let chained = chain_predicate_error(pred_err, err);
171 return fail_after_finally(
172 &mut self.finally,
173 current_ex,
174 chained,
175 )
176 .await;
177 }
178 }
179 }
180 match catch.body.run(current_ex).await {
181 PipelineOutcome::Stopped(stopped_ex) => {
183 return PipelineOutcome::Stopped(stopped_ex);
184 }
185 PipelineOutcome::Completed(next) => {
187 match catch.disposition {
188 ExceptionDisposition::Handled => {
189 current_ex = next;
190 catch_matched = true;
191 break;
192 }
193 ExceptionDisposition::Continued => {
194 current_ex = next;
195 catch_matched = true;
196 break;
197 }
198 _ => {
201 return fail_after_finally(&mut self.finally, next, err)
202 .await;
203 }
204 }
205 }
206 PipelineOutcome::Failed(catch_err) => {
213 tracing::warn!(
220 original_error = %err,
221 catch_error = %catch_err,
222 "do_try catch block failed; catch error supersedes original"
223 );
224 record_span_error(&catch_err);
225 return PipelineOutcome::Failed(catch_err);
226 }
227 }
228 }
229 if catch_matched {
230 current_ex
233 } else {
234 return fail_after_finally(&mut self.finally, current_ex, err).await;
240 }
241 }
242 };
243 match run_finally_body(&mut self.finally, returned_ex).await {
246 Ok(ex) => PipelineOutcome::Completed(ex),
247 Err(FinallyOutcome::Stopped(e)) => PipelineOutcome::Stopped(*e),
248 Err(FinallyOutcome::Failed(finally_err)) => {
249 tracing::warn!(
250 error = %finally_err,
251 "doFinally threw during/after catch; surfacing finally error"
252 );
253 PipelineOutcome::Failed(finally_err)
254 }
255 Err(FinallyOutcome::PredicateFailed(pred_err)) => PipelineOutcome::Failed(pred_err),
258 }
259 })
260 }
261}
262
263#[cfg(test)]
264mod tests {
265 use super::*;
266 use crate::do_try::CatchMatcher;
267 use crate::test_log_capture::{capture_debugs_with_span_records, captured_field, record_field};
268 use camel_api::pipeline_outcome::PipelineOutcome;
269 use std::sync::Arc;
270 use std::sync::atomic::{AtomicU32, Ordering};
271
272 struct CompleteSegment;
275
276 impl OutcomePipeline for CompleteSegment {
277 fn clone_box(&self) -> Box<dyn OutcomePipeline> {
278 Box::new(CompleteSegment)
279 }
280 fn run<'a>(
281 &'a mut self,
282 exchange: Exchange,
283 ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
284 Box::pin(async move { PipelineOutcome::Completed(exchange) })
285 }
286 }
287
288 fn seg_complete() -> camel_api::OutcomeSegment {
289 camel_api::OutcomeSegment::new(Box::new(CompleteSegment))
290 }
291
292 struct FailSegment(CamelError);
293
294 impl OutcomePipeline for FailSegment {
295 fn clone_box(&self) -> Box<dyn OutcomePipeline> {
296 Box::new(FailSegment(self.0.clone()))
297 }
298 fn run<'a>(
299 &'a mut self,
300 _exchange: Exchange,
301 ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
302 let e = self.0.clone();
303 Box::pin(async move { PipelineOutcome::Failed(e) })
304 }
305 }
306
307 fn seg_fail(err: CamelError) -> camel_api::OutcomeSegment {
308 camel_api::OutcomeSegment::new(Box::new(FailSegment(err)))
309 }
310
311 struct MutateThenStop {
312 mutator: Arc<dyn Fn(&mut Exchange) + Send + Sync>,
313 }
314
315 impl OutcomePipeline for MutateThenStop {
316 fn clone_box(&self) -> Box<dyn OutcomePipeline> {
317 Box::new(MutateThenStop {
318 mutator: Arc::clone(&self.mutator),
319 })
320 }
321 fn run<'a>(
322 &'a mut self,
323 mut exchange: Exchange,
324 ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
325 let m = Arc::clone(&self.mutator);
326 Box::pin(async move {
327 m(&mut exchange);
328 PipelineOutcome::Stopped(exchange)
329 })
330 }
331 }
332
333 fn seg_stop_with(
334 mutator: impl Fn(&mut Exchange) + Send + Sync + 'static,
335 ) -> camel_api::OutcomeSegment {
336 camel_api::OutcomeSegment::new(Box::new(MutateThenStop {
337 mutator: Arc::new(mutator),
338 }))
339 }
340
341 struct RecordCall {
342 counter: Arc<AtomicU32>,
343 }
344
345 impl OutcomePipeline for RecordCall {
346 fn clone_box(&self) -> Box<dyn OutcomePipeline> {
347 Box::new(RecordCall {
348 counter: Arc::clone(&self.counter),
349 })
350 }
351 fn run<'a>(
352 &'a mut self,
353 exchange: Exchange,
354 ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
355 let c = Arc::clone(&self.counter);
356 Box::pin(async move {
357 c.fetch_add(1, Ordering::SeqCst);
358 PipelineOutcome::Completed(exchange)
359 })
360 }
361 }
362
363 fn seg_record(counter: Arc<AtomicU32>) -> camel_api::OutcomeSegment {
364 camel_api::OutcomeSegment::new(Box::new(RecordCall { counter }))
365 }
366
367 #[tokio::test]
370 async fn stop_inside_try_skips_catch_and_finally() {
371 let catch_call = Arc::new(AtomicU32::new(0));
372 let finally_call = Arc::new(AtomicU32::new(0));
373
374 let mut seg = DoTrySegment {
375 try_body: seg_stop_with(|ex| {
376 ex.set_property("mutated", camel_api::Value::Bool(true));
377 }),
378 catches: vec![CatchClauseSegment {
379 matcher: CatchMatcher::ByVariant(vec!["*".into()]),
380 on_when: None,
381 body: seg_record(catch_call.clone()),
382 disposition: ExceptionDisposition::Handled,
383 }],
384 finally: Some(FinallyClauseSegment {
385 on_when: None,
386 body: seg_record(finally_call.clone()),
387 }),
388 };
389
390 let result = seg.run(Exchange::default()).await;
391 match result {
392 PipelineOutcome::Stopped(ex) => {
393 assert_eq!(
394 ex.properties.get("mutated"),
395 Some(&camel_api::Value::Bool(true)),
396 "try body mutation must be preserved in Stopped exchange"
397 );
398 }
399 other => panic!("expected Stopped, got {:?}", other),
400 }
401 assert_eq!(
402 catch_call.load(Ordering::SeqCst),
403 0,
404 "catch must NOT run when try stops"
405 );
406 assert_eq!(
407 finally_call.load(Ordering::SeqCst),
408 0,
409 "finally must NOT run when try stops"
410 );
411 }
412
413 #[tokio::test]
414 async fn stop_inside_catch_skips_finally() {
415 let finally_call = Arc::new(AtomicU32::new(0));
416
417 let mut seg = DoTrySegment {
418 try_body: seg_fail(CamelError::ProcessorError("boom".into())),
419 catches: vec![CatchClauseSegment {
420 matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
421 on_when: None,
422 body: seg_stop_with(|ex| {
423 ex.set_property("catch_mutated", camel_api::Value::Bool(true));
424 }),
425 disposition: ExceptionDisposition::Handled,
426 }],
427 finally: Some(FinallyClauseSegment {
428 on_when: None,
429 body: seg_record(finally_call.clone()),
430 }),
431 };
432
433 let result = seg.run(Exchange::default()).await;
434 match result {
435 PipelineOutcome::Stopped(ex) => {
436 assert_eq!(
437 ex.properties.get("catch_mutated"),
438 Some(&camel_api::Value::Bool(true)),
439 "catch body mutation must be preserved in Stopped exchange"
440 );
441 }
442 other => panic!("expected Stopped, got {:?}", other),
443 }
444 assert_eq!(
445 finally_call.load(Ordering::SeqCst),
446 0,
447 "finally must NOT run when catch stops"
448 );
449 }
450
451 #[tokio::test]
452 async fn stop_inside_finally_stops_outer_route() {
453 let mut seg = DoTrySegment {
454 try_body: seg_complete(),
455 catches: vec![],
456 finally: Some(FinallyClauseSegment {
457 on_when: None,
458 body: seg_stop_with(|ex| {
459 ex.set_property("finally_mutated", camel_api::Value::Bool(true));
460 }),
461 }),
462 };
463
464 let result = seg.run(Exchange::default()).await;
465 match result {
466 PipelineOutcome::Stopped(ex) => {
467 assert_eq!(
468 ex.properties.get("finally_mutated"),
469 Some(&camel_api::Value::Bool(true)),
470 "finally body mutation must be preserved in Stopped exchange"
471 );
472 }
473 other => panic!("expected Stopped, got {:?}", other),
474 }
475 }
476
477 #[tokio::test]
478 async fn catch_on_when_false_falls_through_to_next_catch() {
479 let first_call = Arc::new(AtomicU32::new(0));
480 let second_call = Arc::new(AtomicU32::new(0));
481
482 let mut seg = DoTrySegment {
483 try_body: seg_fail(CamelError::Io("disk err".into())),
484 catches: vec![
485 CatchClauseSegment {
486 matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
487 on_when: Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false))),
488 body: seg_record(first_call.clone()),
489 disposition: ExceptionDisposition::Handled,
490 },
491 CatchClauseSegment {
492 matcher: CatchMatcher::ByVariant(vec!["*".into()]),
493 on_when: None,
494 body: seg_record(second_call.clone()),
495 disposition: ExceptionDisposition::Handled,
496 },
497 ],
498 finally: None,
499 };
500
501 let result = seg.run(Exchange::default()).await;
502 assert!(
503 matches!(result, PipelineOutcome::Completed(_)),
504 "expected Completed after second catch"
505 );
506 assert_eq!(
507 first_call.load(Ordering::SeqCst),
508 0,
509 "first catch must NOT fire (on_when=false)"
510 );
511 assert_eq!(
512 second_call.load(Ordering::SeqCst),
513 1,
514 "second catch must fire"
515 );
516 }
517
518 #[tokio::test]
519 async fn finally_on_when_false_skips_finally_entirely() {
520 let finally_call = Arc::new(AtomicU32::new(0));
521 let mut ex = Exchange::default();
522 ex.set_property("try_set", camel_api::Value::Bool(true));
523
524 let mut seg = DoTrySegment {
525 try_body: seg_complete(),
526 catches: vec![],
527 finally: Some(FinallyClauseSegment {
528 on_when: Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false))),
529 body: seg_record(finally_call.clone()),
530 }),
531 };
532
533 let result = seg.run(ex).await;
534 match result {
535 PipelineOutcome::Completed(ex) => {
536 assert_eq!(
537 ex.properties.get("try_set"),
538 Some(&camel_api::Value::Bool(true)),
539 "exchange state from try must be preserved"
540 );
541 }
542 other => panic!("expected Completed, got {:?}", other),
543 }
544 assert_eq!(
545 finally_call.load(Ordering::SeqCst),
546 0,
547 "finally must NOT run when on_when=false"
548 );
549 }
550
551 fn catch_fails_segment() -> DoTrySegment {
554 DoTrySegment {
555 try_body: seg_fail(CamelError::ProcessorError("orig".into())),
556 catches: vec![CatchClauseSegment {
557 matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
558 on_when: None,
559 body: seg_fail(CamelError::Io("catch-fail".into())),
560 disposition: ExceptionDisposition::Handled,
561 }],
562 finally: None,
563 }
564 }
565
566 #[test]
567 fn catch_body_failure_returns_catch_err_and_marks_span() {
568 let mut seg = catch_fails_segment();
569 let (result, _captured, span_records) = capture_debugs_with_span_records(|| {
570 let span = tracing::info_span!("dotry_seg_test", error = tracing::field::Empty);
572 let _guard = span.enter();
573 tokio::runtime::Builder::new_current_thread()
574 .enable_all()
575 .build()
576 .expect("current-thread runtime")
577 .block_on(async { seg.run(Exchange::default()).await })
578 });
579
580 assert!(
581 matches!(result, PipelineOutcome::Failed(CamelError::Io(_))),
582 "catch failure must surface the catch error as Failed(Io), got: {result:?}"
583 );
584 assert!(
588 span_records.iter().any(|line| {
589 captured_field(line, "error").is_some_and(|v| v.contains("catch-fail"))
590 }),
591 "expected span error record carrying the catch error, span records: {span_records:?}"
592 );
593 }
594
595 #[test]
596 fn catch_body_failure_emits_envelope_log() {
597 let mut seg = catch_fails_segment();
598 let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
599 tokio::runtime::Builder::new_current_thread()
600 .enable_all()
601 .build()
602 .expect("current-thread runtime")
603 .block_on(async { seg.run(Exchange::default()).await })
604 });
605
606 assert!(
607 matches!(result, PipelineOutcome::Failed(CamelError::Io(_))),
608 "catch failure must surface the catch error as Failed(Io), got: {result:?}"
609 );
610 let original = record_field(&captured, "do_try catch block failed", "original_error")
613 .unwrap_or_else(|| {
614 panic!("envelope record missing original_error field, captured: {captured:?}")
615 });
616 let catch = record_field(&captured, "do_try catch block failed", "catch_error")
617 .unwrap_or_else(|| {
618 panic!("envelope record missing catch_error field, captured: {captured:?}")
619 });
620 assert!(
621 original.contains("orig"),
622 "original_error must carry the original error, got: {original}"
623 );
624 assert!(
625 catch.contains("catch-fail"),
626 "catch_error must carry the catch error, got: {catch}"
627 );
628 }
629
630 use camel_api::{ExpressionErrorClass, FilterPredicate, PredicateSource};
633
634 fn expression_failed() -> CamelError {
635 CamelError::ExpressionFailed {
636 language: "rhai".to_string(),
637 route_id: "r1".to_string(),
638 step_id: "step#0".to_string(),
639 verb: "on_when".to_string(),
640 class: ExpressionErrorClass::Runtime,
641 position: None,
642 conversion: None,
643 cause: None,
644 }
645 }
646
647 fn async_err_predicate(err: CamelError) -> PredicateSource {
648 PredicateSource::Async(Arc::new(move |_: &Exchange| {
649 let err = err.clone();
650 Box::pin(async move { Err(err) }) as camel_api::BoxBoolFuture
651 }))
652 }
653
654 #[tokio::test]
655 async fn finally_on_when_predicate_error_fails_segment() {
656 let finally_call = Arc::new(AtomicU32::new(0));
657 let mut seg = DoTrySegment {
658 try_body: seg_complete(),
659 catches: vec![],
660 finally: Some(FinallyClauseSegment {
661 on_when: Some(async_err_predicate(expression_failed())),
662 body: seg_record(finally_call.clone()),
663 }),
664 };
665
666 let result = seg.run(Exchange::default()).await;
667 match result {
668 PipelineOutcome::Failed(CamelError::ExpressionFailed { .. }) => {}
669 other => panic!("expected Failed(ExpressionFailed), got {other:?}"),
670 }
671 assert_eq!(
672 finally_call.load(Ordering::SeqCst),
673 0,
674 "finally body must not run when its on_when predicate errors"
675 );
676 }
677
678 #[tokio::test]
685 async fn no_clause_matches_runs_finally_and_propagates_original() {
686 let catch_call = Arc::new(AtomicU32::new(0));
687 let finally_call = Arc::new(AtomicU32::new(0));
688
689 let mut seg = DoTrySegment {
690 try_body: seg_fail(CamelError::EndpointCreationFailed(
691 "direct endpoint 'direct:does-not-exist' not registered".into(),
692 )),
693 catches: vec![CatchClauseSegment {
694 matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
695 on_when: None,
696 body: seg_record(catch_call.clone()),
697 disposition: ExceptionDisposition::Handled,
698 }],
699 finally: Some(FinallyClauseSegment {
700 on_when: None,
701 body: seg_record(finally_call.clone()),
702 }),
703 };
704
705 let result = seg.run(Exchange::default()).await;
706 match result {
707 PipelineOutcome::Failed(CamelError::EndpointCreationFailed(msg)) => {
708 assert!(
709 msg.contains("does-not-exist"),
710 "propagated error must be the original try error, got: {msg}"
711 );
712 }
713 other => {
714 panic!("no catch matched: expected Failed(EndpointCreationFailed), got {other:?}")
715 }
716 }
717 assert_eq!(
718 finally_call.load(Ordering::SeqCst),
719 1,
720 "finally must run exactly once when no catch matches"
721 );
722 assert_eq!(
723 catch_call.load(Ordering::SeqCst),
724 0,
725 "the unmatched catch body must not run"
726 );
727 }
728}