1use std::sync::Arc;
5
6use camel_api::UnitOfWorkConfig;
7use camel_api::circuit_breaker::CircuitBreakerConfig;
8use camel_api::error_handler::ErrorHandlerConfig;
9use camel_api::loop_eip::LoopConfig;
10use camel_api::security_policy::SecurityPolicyConfig;
11use camel_api::{
12 AggregatorConfig, FilterPredicate, MulticastConfig, OpaqueProcessor, ResequencePolicyConfig,
13 SplitterConfig,
14};
15use camel_auth::TokenAuthenticator;
16use camel_component_api::ConcurrencyModel;
17
18#[derive(Clone)]
20pub struct WhenStep {
21 pub predicate: FilterPredicate,
22 pub steps: Vec<BuilderStep>,
23}
24
25impl std::fmt::Debug for WhenStep {
26 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27 f.debug_struct("WhenStep")
28 .field("predicate", &self.predicate)
29 .field("steps", &self.steps)
30 .finish()
31 }
32}
33
34pub use camel_api::declarative::{LanguageExpressionDef, ValueSourceDef};
35
36#[derive(Debug, Clone)]
38pub struct DeclarativeWhenStep {
39 pub predicate: LanguageExpressionDef,
40 pub steps: Vec<BuilderStep>,
41}
42
43#[derive(Debug, Clone)]
45pub struct DoTryCatchClauseBuilder {
46 pub exception: Option<Vec<String>>,
47 pub when: Option<LanguageExpressionDef>,
48 pub on_when: Option<LanguageExpressionDef>,
49 pub disposition: camel_api::error_handler::ExceptionDisposition,
50 pub steps: Vec<BuilderStep>,
51}
52
53#[derive(Debug, Clone)]
55pub struct DoTryFinallyBuilder {
56 pub on_when: Option<LanguageExpressionDef>,
57 pub steps: Vec<BuilderStep>,
58}
59
60#[derive(Debug, Clone)]
62pub enum BuilderStep {
63 Processor(OpaqueProcessor),
65 To(String),
67 Stop,
69 Log {
71 level: camel_processor::LogLevel,
72 message: String,
73 },
74 DeclarativeSetHeader {
76 key: String,
77 value: ValueSourceDef,
78 },
79 DeclarativeSetHeaderIfAbsent {
81 key: String,
82 value: ValueSourceDef,
83 },
84 DeclarativeRemoveHeader {
86 key: String,
87 },
88 DeclarativeSetProperty {
89 key: String,
90 value_source: ValueSourceDef,
91 },
92 DeclarativeSetBody {
94 value: ValueSourceDef,
95 },
96 DeclarativeFilter {
98 predicate: LanguageExpressionDef,
99 steps: Vec<BuilderStep>,
100 },
101 DeclarativeChoice {
103 whens: Vec<DeclarativeWhenStep>,
104 otherwise: Option<Vec<BuilderStep>>,
105 },
106 DeclarativeScript {
108 expression: LanguageExpressionDef,
109 },
110 DeclarativeFunction {
111 definition: camel_api::FunctionDefinition,
112 },
113 DeclarativeSplit {
115 expression: LanguageExpressionDef,
116 aggregation: camel_api::splitter::AggregationStrategy,
117 parallel: bool,
118 parallel_limit: Option<usize>,
119 stop_on_exception: bool,
120 steps: Vec<BuilderStep>,
121 },
122 DeclarativeStreamSplit {
124 stream_config: camel_api::StreamSplitConfig,
125 aggregation: camel_api::splitter::AggregationStrategy,
126 stop_on_exception: bool,
127 steps: Vec<BuilderStep>,
128 },
129 DeclarativeDynamicRouter {
130 expression: LanguageExpressionDef,
131 uri_delimiter: String,
132 cache_size: i32,
133 ignore_invalid_endpoints: bool,
134 max_iterations: usize,
135 },
136 DeclarativeRoutingSlip {
137 expression: LanguageExpressionDef,
138 uri_delimiter: String,
139 cache_size: i32,
140 ignore_invalid_endpoints: bool,
141 },
142 Split {
144 config: SplitterConfig,
145 steps: Vec<BuilderStep>,
146 },
147 Aggregate {
149 config: AggregatorConfig,
150 },
151 Filter {
153 predicate: FilterPredicate,
154 steps: Vec<BuilderStep>,
155 },
156 Choice {
159 whens: Vec<WhenStep>,
160 otherwise: Option<Vec<BuilderStep>>,
161 },
162 WireTap {
164 uri: String,
165 },
166 Multicast {
168 steps: Vec<BuilderStep>,
169 config: MulticastConfig,
170 },
171 DeclarativeLog {
173 level: camel_processor::LogLevel,
174 message: ValueSourceDef,
175 },
176 Bean {
178 name: String,
179 method: String,
180 },
181 Script {
184 language: String,
185 script: String,
186 },
187 Throttle {
189 config: camel_api::ThrottlerConfig,
190 steps: Vec<BuilderStep>,
191 },
192 LoadBalance {
194 config: camel_api::LoadBalancerConfig,
195 steps: Vec<BuilderStep>,
196 },
197 DynamicRouter {
199 config: camel_api::DynamicRouterConfig,
200 },
201 RoutingSlip {
202 config: camel_api::RoutingSlipConfig,
203 },
204 RecipientList {
205 config: camel_api::recipient_list::RecipientListConfig,
206 },
207 DeclarativeRecipientList {
208 expression: LanguageExpressionDef,
209 delimiter: String,
210 parallel: bool,
211 parallel_limit: Option<usize>,
212 stop_on_exception: bool,
213 aggregation: String,
214 },
215 Delay {
216 config: camel_api::DelayConfig,
217 },
218 Loop {
220 config: LoopConfig,
221 steps: Vec<BuilderStep>,
222 },
223 DeclarativeLoop {
225 count: Option<usize>,
226 while_predicate: Option<LanguageExpressionDef>,
227 steps: Vec<BuilderStep>,
228 max_iterations: Option<usize>,
229 },
230 Enrich {
232 uri: String,
233 strategy: Option<String>,
234 timeout_ms: Option<u64>,
235 },
236 PollEnrich {
238 uri: String,
239 strategy: Option<String>,
240 timeout_ms: Option<u64>,
241 },
242 Validate {
245 predicate: LanguageExpressionDef,
246 },
247 ClaimCheck {
251 repository: String,
252 operation: String,
253 key: LanguageExpressionDef,
254 filter: Option<String>,
255 },
256 Sampling {
260 period: usize,
261 },
262 Sort {
265 expression: LanguageExpressionDef,
266 reverse: bool,
267 },
268 IdempotentConsumer {
272 repository: String,
273 expression: LanguageExpressionDef,
274 steps: Vec<BuilderStep>,
275 eager: bool,
276 remove_on_failure: bool,
277 },
278 Cache {
282 repository: Option<String>,
283 key: LanguageExpressionDef,
284 ttl: Option<String>,
285 max_entry_bytes: Option<usize>,
286 coalesce_misses: bool,
288 on_miss: Vec<BuilderStep>,
289 },
290 CacheInvalidate {
295 repository: Option<String>,
296 key: Option<LanguageExpressionDef>,
297 key_prefix: Option<LanguageExpressionDef>,
298 },
299 CacheClear {
302 repository: Option<String>,
303 },
304 CacheStats {
307 repository: Option<String>,
308 },
309 CachePeekStale {
312 repository: Option<String>,
313 key: LanguageExpressionDef,
314 on_miss: camel_processor::PeekStaleMissPolicy,
315 },
316 DeclarativeDoTry {
318 try_steps: Vec<BuilderStep>,
319 catch: Vec<DoTryCatchClauseBuilder>,
320 finally: Option<DoTryFinallyBuilder>,
321 },
322 Resequence {
325 policy_config: ResequencePolicyConfig,
326 },
327}
328
329pub struct RouteDefinition {
331 pub(crate) from_uri: String,
332 pub(crate) steps: Vec<BuilderStep>,
333 pub(crate) error_handler: Option<ErrorHandlerConfig>,
335 pub(crate) circuit_breaker: Option<CircuitBreakerConfig>,
337 pub(crate) circuit_breaker_fallback: Vec<BuilderStep>,
345 pub(crate) security_policy: Option<SecurityPolicyConfig>,
346 pub(crate) security_authenticator: Option<Arc<dyn TokenAuthenticator>>,
348 pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
350 pub(crate) concurrency: Option<ConcurrencyModel>,
353 pub(crate) route_id: String,
355 pub(crate) auto_startup: bool,
357 pub(crate) startup_order: i32,
359 pub(crate) source_hash: Option<u64>,
360}
361
362impl RouteDefinition {
363 pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
365 Self {
366 from_uri: from_uri.into(),
367 steps,
368 error_handler: None,
369 circuit_breaker: None,
370 circuit_breaker_fallback: Vec::new(),
371 security_policy: None,
372 security_authenticator: None,
373 unit_of_work: None,
374 concurrency: None,
375 route_id: String::new(), auto_startup: true,
377 startup_order: 1000,
378 source_hash: None,
379 }
380 }
381
382 pub fn from_uri(&self) -> &str {
384 &self.from_uri
385 }
386
387 pub fn steps(&self) -> &[BuilderStep] {
389 &self.steps
390 }
391
392 pub fn map_steps(mut self, f: impl FnOnce(Vec<BuilderStep>) -> Vec<BuilderStep>) -> Self {
397 self.steps = f(self.steps);
398 self
399 }
400
401 pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
403 self.error_handler = Some(config);
404 self
405 }
406
407 pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
409 self.error_handler.as_ref()
410 }
411
412 pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
414 self.circuit_breaker = Some(config);
415 self
416 }
417
418 pub fn with_circuit_breaker_fallback(mut self, steps: Vec<BuilderStep>) -> Self {
425 self.circuit_breaker_fallback = steps;
426 self
427 }
428
429 pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
431 self.security_policy = Some(config);
432 self
433 }
434
435 pub fn with_security_authenticator(
437 mut self,
438 authenticator: Arc<dyn TokenAuthenticator>,
439 ) -> Self {
440 self.security_authenticator = Some(authenticator);
441 self
442 }
443
444 pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
446 self.unit_of_work = Some(config);
447 self
448 }
449
450 pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
452 self.unit_of_work.as_ref()
453 }
454
455 pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
457 self.circuit_breaker.as_ref()
458 }
459
460 pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
461 self.security_policy.as_ref()
462 }
463
464 pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
465 self.security_authenticator.as_ref()
466 }
467
468 pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
470 self.concurrency.as_ref()
471 }
472
473 pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
475 self.concurrency = Some(model);
476 self
477 }
478
479 pub fn route_id(&self) -> &str {
481 &self.route_id
482 }
483
484 pub fn auto_startup(&self) -> bool {
486 self.auto_startup
487 }
488
489 pub fn startup_order(&self) -> i32 {
491 self.startup_order
492 }
493
494 pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
496 self.route_id = id.into();
497 self
498 }
499
500 pub fn with_auto_startup(mut self, auto: bool) -> Self {
502 self.auto_startup = auto;
503 self
504 }
505
506 pub fn with_startup_order(mut self, order: i32) -> Self {
508 self.startup_order = order;
509 self
510 }
511
512 pub fn with_source_hash(mut self, hash: u64) -> Self {
513 self.source_hash = Some(hash);
514 self
515 }
516
517 pub fn source_hash(&self) -> Option<u64> {
518 self.source_hash
519 }
520
521 pub fn to_info(&self) -> RouteDefinitionInfo {
524 RouteDefinitionInfo {
525 route_id: self.route_id.clone(),
526 auto_startup: self.auto_startup,
527 startup_order: self.startup_order,
528 source_hash: self.source_hash,
529 }
530 }
531}
532
533#[derive(Clone)]
539pub struct RouteDefinitionInfo {
540 route_id: String,
541 auto_startup: bool,
542 startup_order: i32,
543 pub(crate) source_hash: Option<u64>,
544}
545
546impl RouteDefinitionInfo {
547 pub fn route_id(&self) -> &str {
549 &self.route_id
550 }
551
552 pub fn auto_startup(&self) -> bool {
554 self.auto_startup
555 }
556
557 pub fn startup_order(&self) -> i32 {
559 self.startup_order
560 }
561
562 pub fn source_hash(&self) -> Option<u64> {
563 self.source_hash
564 }
565}
566
567#[cfg(test)]
568mod tests {
569 use super::*;
570
571 #[test]
576 fn golden_debug_output_all_variants() {
577 use camel_api::declarative::LanguageExpressionDef;
578 use camel_api::loop_eip::LoopMode;
579 use camel_api::recipient_list::RecipientListConfig;
580 use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
581 use camel_api::{
582 BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
583 FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
584 Value,
585 };
586 use std::sync::Arc;
587
588 let expr = LanguageExpressionDef {
589 language: "simple".into(),
590 source: "${body}".into(),
591 };
592
593 assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
596 assert_eq!(
597 format!(
598 "{:?}",
599 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
600 ),
601 "Processor(BoxProcessor(...))"
602 );
603 assert_eq!(
604 format!("{:?}", BuilderStep::To("mock:out".into())),
605 "To(\"mock:out\")"
606 );
607
608 assert_eq!(
611 format!(
612 "{:?}",
613 BuilderStep::Log {
614 level: camel_processor::LogLevel::Info,
615 message: "hello".into(),
616 }
617 ),
618 "Log { level: Info, message: \"hello\" }"
619 );
620
621 assert_eq!(
622 format!(
623 "{:?}",
624 BuilderStep::DeclarativeSetHeader {
625 key: "k".into(),
626 value: ValueSourceDef::Literal(Value::String("v".into())),
627 }
628 ),
629 "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
630 );
631
632 assert_eq!(
633 format!(
634 "{:?}",
635 BuilderStep::DeclarativeSetHeaderIfAbsent {
636 key: "k".into(),
637 value: ValueSourceDef::Literal(Value::String("v".into())),
638 }
639 ),
640 "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
641 );
642
643 assert_eq!(
644 format!(
645 "{:?}",
646 BuilderStep::DeclarativeSetBody {
647 value: ValueSourceDef::Literal(Value::String("v".into())),
648 }
649 ),
650 "DeclarativeSetBody { value: Literal(String(\"v\")) }"
651 );
652
653 assert_eq!(
654 format!(
655 "{:?}",
656 BuilderStep::DeclarativeSetProperty {
657 key: "prop".into(),
658 value_source: ValueSourceDef::Literal(Value::String("v".into())),
659 }
660 ),
661 "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
662 );
663
664 assert_eq!(
665 format!(
666 "{:?}",
667 BuilderStep::DeclarativeScript {
668 expression: expr.clone(),
669 }
670 ),
671 "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
672 );
673
674 let func_def = FunctionDefinition {
676 id: FunctionId("test-id".into()),
677 runtime: "my_runtime".into(),
678 source: "${body}".into(),
679 timeout_ms: 5000,
680 route_id: None,
681 step_index: None,
682 };
683 assert_eq!(
684 format!(
685 "{:?}",
686 BuilderStep::DeclarativeFunction {
687 definition: func_def,
688 }
689 ),
690 "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
691 );
692
693 assert_eq!(
694 format!(
695 "{:?}",
696 BuilderStep::WireTap {
697 uri: "mock:tap".into(),
698 }
699 ),
700 "WireTap { uri: \"mock:tap\" }"
701 );
702
703 assert_eq!(
704 format!(
705 "{:?}",
706 BuilderStep::DeclarativeLog {
707 level: camel_processor::LogLevel::Info,
708 message: ValueSourceDef::Expression(expr.clone()),
709 }
710 ),
711 "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
712 );
713
714 assert_eq!(
715 format!(
716 "{:?}",
717 BuilderStep::Bean {
718 name: "myBean".into(),
719 method: "process".into(),
720 }
721 ),
722 "Bean { name: \"myBean\", method: \"process\" }"
723 );
724
725 assert_eq!(
726 format!(
727 "{:?}",
728 BuilderStep::Script {
729 language: "js".into(),
730 script: "body".into(),
731 }
732 ),
733 "Script { language: \"js\", script: \"body\" }"
734 );
735
736 assert_eq!(
737 format!(
738 "{:?}",
739 BuilderStep::Aggregate {
740 config: camel_api::AggregatorConfig::correlate_by("id")
741 .complete_when_size(1)
742 .build()
743 .unwrap(),
744 }
745 ),
746 "Aggregate { config: AggregatorConfig { header_name: \"id\", completion: Single(Size(1)), correlation: HeaderName(\"id\"), strategy: CollectAll, max_buckets: Some(10000), bucket_ttl: Some(300s), force_completion_on_stop: false, discard_on_timeout: false, max_timeout_tasks: 1024 } }"
747 );
748
749 assert_eq!(
750 format!(
751 "{:?}",
752 BuilderStep::DynamicRouter {
753 config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
754 "mock:dr".into()
755 ))),
756 }
757 ),
758 "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
759 );
760
761 assert_eq!(
762 format!(
763 "{:?}",
764 BuilderStep::RoutingSlip {
765 config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
766 }
767 ),
768 "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
769 );
770
771 assert_eq!(
772 format!(
773 "{:?}",
774 BuilderStep::RecipientList {
775 config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
776 }
777 ),
778 "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
779 );
780
781 assert_eq!(
782 format!(
783 "{:?}",
784 BuilderStep::Enrich {
785 uri: "mock:enrich".into(),
786 strategy: Some("agg".into()),
787 timeout_ms: Some(1000),
788 }
789 ),
790 "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
791 );
792
793 assert_eq!(
794 format!(
795 "{:?}",
796 BuilderStep::PollEnrich {
797 uri: "mock:poll".into(),
798 strategy: None,
799 timeout_ms: None,
800 }
801 ),
802 "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
803 );
804
805 assert_eq!(
806 format!(
807 "{:?}",
808 BuilderStep::Validate {
809 predicate: expr.clone(),
810 }
811 ),
812 "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
813 );
814
815 assert_eq!(
816 format!("{:?}", BuilderStep::Sampling { period: 100 }),
817 "Sampling { period: 100 }"
818 );
819
820 assert_eq!(
821 format!(
822 "{:?}",
823 BuilderStep::Resequence {
824 policy_config: Default::default(),
825 }
826 ),
827 "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
828 );
829
830 assert_eq!(
832 format!(
833 "{:?}",
834 BuilderStep::DeclarativeFilter {
835 predicate: expr.clone(),
836 steps: vec![BuilderStep::Stop],
837 }
838 ),
839 "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
840 );
841
842 assert_eq!(
843 format!(
844 "{:?}",
845 BuilderStep::DeclarativeSplit {
846 expression: expr.clone(),
847 aggregation: AggregationStrategy::Original,
848 parallel: false,
849 parallel_limit: Some(2),
850 stop_on_exception: true,
851 steps: vec![BuilderStep::Stop],
852 }
853 ),
854 "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
855 );
856
857 assert_eq!(
858 format!(
859 "{:?}",
860 BuilderStep::Split {
861 config: camel_api::splitter::SplitterConfig::new(
862 camel_api::splitter::split_body_lines()
863 ),
864 steps: vec![BuilderStep::Stop],
865 }
866 ),
867 "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
868 );
869
870 assert_eq!(
871 format!(
872 "{:?}",
873 BuilderStep::Filter {
874 predicate: FilterPredicate::new(|_: &Exchange| true),
875 steps: vec![BuilderStep::Stop],
876 }
877 ),
878 "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
879 );
880
881 assert_eq!(
882 format!(
883 "{:?}",
884 BuilderStep::Throttle {
885 config: camel_api::ThrottlerConfig::new(
886 10,
887 std::time::Duration::from_millis(10)
888 ),
889 steps: vec![BuilderStep::Stop],
890 }
891 ),
892 "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
893 );
894
895 assert_eq!(
896 format!(
897 "{:?}",
898 BuilderStep::LoadBalance {
899 config: camel_api::LoadBalancerConfig::round_robin(),
900 steps: vec![BuilderStep::To("mock:l1".into())],
901 }
902 ),
903 "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
904 );
905
906 assert_eq!(
907 format!(
908 "{:?}",
909 BuilderStep::Delay {
910 config: camel_api::DelayConfig::new(500),
911 }
912 ),
913 "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
914 );
915
916 assert_eq!(
920 format!(
921 "{:?}",
922 BuilderStep::Choice {
923 whens: vec![WhenStep {
924 predicate: FilterPredicate::new(|_: &Exchange| true),
925 steps: vec![BuilderStep::To("mock:a".into())],
926 }],
927 otherwise: None,
928 }
929 ),
930 "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
931 );
932
933 assert_eq!(
934 format!(
935 "{:?}",
936 BuilderStep::DeclarativeChoice {
937 whens: vec![DeclarativeWhenStep {
938 predicate: expr.clone(),
939 steps: vec![BuilderStep::Stop],
940 }],
941 otherwise: Some(vec![BuilderStep::Stop]),
942 }
943 ),
944 "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
945 );
946
947 assert_eq!(
949 format!(
950 "{:?}",
951 BuilderStep::Multicast {
952 steps: vec![BuilderStep::To("direct:a".into())],
953 config: MulticastConfig::new(),
954 }
955 ),
956 "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
957 );
958
959 assert_eq!(
961 format!(
962 "{:?}",
963 BuilderStep::DeclarativeDynamicRouter {
964 expression: expr.clone(),
965 uri_delimiter: ",".into(),
966 cache_size: 1000,
967 ignore_invalid_endpoints: false,
968 max_iterations: 1000,
969 }
970 ),
971 "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
972 );
973
974 assert_eq!(
975 format!(
976 "{:?}",
977 BuilderStep::DeclarativeRoutingSlip {
978 expression: expr.clone(),
979 uri_delimiter: ",".into(),
980 cache_size: 1000,
981 ignore_invalid_endpoints: false,
982 }
983 ),
984 "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
985 );
986
987 assert_eq!(
989 format!(
990 "{:?}",
991 BuilderStep::DeclarativeRecipientList {
992 expression: expr.clone(),
993 delimiter: ",".into(),
994 parallel: false,
995 parallel_limit: None,
996 stop_on_exception: false,
997 aggregation: "original".into(),
998 }
999 ),
1000 "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
1001 );
1002
1003 assert_eq!(
1005 format!(
1006 "{:?}",
1007 BuilderStep::Loop {
1008 config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
1009 steps: vec![],
1010 }
1011 ),
1012 "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
1013 );
1014
1015 assert_eq!(
1016 format!(
1017 "{:?}",
1018 BuilderStep::DeclarativeLoop {
1019 count: Some(5),
1020 while_predicate: None,
1021 steps: vec![],
1022 max_iterations: Some(100),
1023 }
1024 ),
1025 "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
1026 );
1027
1028 assert_eq!(
1030 format!(
1031 "{:?}",
1032 BuilderStep::ClaimCheck {
1033 repository: "myRepo".into(),
1034 operation: "checkout".into(),
1035 key: expr.clone(),
1036 filter: None,
1037 }
1038 ),
1039 "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
1040 );
1041
1042 assert_eq!(
1044 format!(
1045 "{:?}",
1046 BuilderStep::Sort {
1047 expression: expr.clone(),
1048 reverse: false,
1049 }
1050 ),
1051 "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
1052 );
1053
1054 assert_eq!(
1056 format!(
1057 "{:?}",
1058 BuilderStep::IdempotentConsumer {
1059 repository: "myRepo".into(),
1060 expression: expr.clone(),
1061 steps: vec![],
1062 eager: true,
1063 remove_on_failure: false,
1064 }
1065 ),
1066 "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
1067 );
1068
1069 assert_eq!(
1071 format!(
1072 "{:?}",
1073 BuilderStep::DeclarativeDoTry {
1074 try_steps: vec![BuilderStep::Stop],
1075 catch: vec![],
1076 finally: None,
1077 }
1078 ),
1079 "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1080 );
1081
1082 assert_eq!(
1084 format!(
1085 "{:?}",
1086 BuilderStep::DeclarativeStreamSplit {
1087 stream_config: StreamSplitConfig {
1088 format: StreamSplitFormat::Ndjson,
1089 max_record_bytes: 1024 * 1024,
1090 batch_size: 1,
1091 chunk_size: None,
1092 include_origin: true,
1093 },
1094 aggregation: AggregationStrategy::Original,
1095 stop_on_exception: true,
1096 steps: vec![BuilderStep::Stop],
1097 }
1098 ),
1099 "DeclarativeStreamSplit { stream_config: StreamSplitConfig { format: Ndjson, max_record_bytes: 1048576, batch_size: 1, chunk_size: None, include_origin: true }, aggregation: Original, stop_on_exception: true, steps: [Stop] }"
1100 );
1101 }
1102
1103 #[test]
1104 fn test_builder_step_multicast_variant() {
1105 use camel_api::MulticastConfig;
1106
1107 let step = BuilderStep::Multicast {
1108 steps: vec![BuilderStep::To("direct:a".into())],
1109 config: MulticastConfig::new(),
1110 };
1111
1112 assert!(matches!(step, BuilderStep::Multicast { .. }));
1113 }
1114
1115 #[test]
1116 fn test_route_definition_defaults() {
1117 let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1118 assert_eq!(def.route_id(), "test-route");
1119 assert!(def.auto_startup());
1120 assert_eq!(def.startup_order(), 1000);
1121 }
1122
1123 #[test]
1124 fn test_route_definition_builders() {
1125 let def = RouteDefinition::new("direct:test", vec![])
1126 .with_route_id("my-route")
1127 .with_auto_startup(false)
1128 .with_startup_order(50);
1129 assert_eq!(def.route_id(), "my-route");
1130 assert!(!def.auto_startup());
1131 assert_eq!(def.startup_order(), 50);
1132 }
1133
1134 #[test]
1135 fn test_route_definition_accessors_cover_core_fields() {
1136 let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1137 .with_route_id("accessor-route");
1138
1139 assert_eq!(def.from_uri(), "direct:in");
1140 assert_eq!(def.steps().len(), 1);
1141 assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1142 }
1143
1144 #[test]
1145 fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1146 use camel_api::circuit_breaker::CircuitBreakerConfig;
1147 use camel_api::error_handler::ErrorHandlerConfig;
1148 use camel_component_api::ConcurrencyModel;
1149
1150 let def = RouteDefinition::new("direct:test", vec![])
1151 .with_route_id("eh-route")
1152 .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1153 .with_circuit_breaker(CircuitBreakerConfig::new())
1154 .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1155
1156 let eh = def
1157 .error_handler_config()
1158 .expect("error handler should be set");
1159 assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1160 assert!(def.circuit_breaker_config().is_some());
1161 assert!(matches!(
1162 def.concurrency_override(),
1163 Some(ConcurrencyModel::Concurrent { max: Some(4) })
1164 ));
1165 }
1166
1167 #[test]
1168 fn test_builder_step_debug_covers_many_variants() {
1169 use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1170 use camel_api::{
1171 BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1172 OpaqueProcessor, RoutingSlipConfig, Value,
1173 };
1174 use std::sync::Arc;
1175
1176 let expr = LanguageExpressionDef {
1177 language: "simple".into(),
1178 source: "${body}".into(),
1179 };
1180
1181 let steps = vec![
1182 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1183 BuilderStep::To("mock:out".into()),
1184 BuilderStep::Stop,
1185 BuilderStep::Log {
1186 level: camel_processor::LogLevel::Info,
1187 message: "hello".into(),
1188 },
1189 BuilderStep::DeclarativeSetHeader {
1190 key: "k".into(),
1191 value: ValueSourceDef::Literal(Value::String("v".into())),
1192 },
1193 BuilderStep::DeclarativeSetBody {
1194 value: ValueSourceDef::Expression(expr.clone()),
1195 },
1196 BuilderStep::DeclarativeFilter {
1197 predicate: expr.clone(),
1198 steps: vec![BuilderStep::Stop],
1199 },
1200 BuilderStep::DeclarativeChoice {
1201 whens: vec![DeclarativeWhenStep {
1202 predicate: expr.clone(),
1203 steps: vec![BuilderStep::Stop],
1204 }],
1205 otherwise: Some(vec![BuilderStep::Stop]),
1206 },
1207 BuilderStep::DeclarativeScript {
1208 expression: expr.clone(),
1209 },
1210 BuilderStep::DeclarativeSplit {
1211 expression: expr.clone(),
1212 aggregation: AggregationStrategy::Original,
1213 parallel: false,
1214 parallel_limit: Some(2),
1215 stop_on_exception: true,
1216 steps: vec![BuilderStep::Stop],
1217 },
1218 BuilderStep::Split {
1219 config: SplitterConfig::new(split_body_lines()),
1220 steps: vec![BuilderStep::Stop],
1221 },
1222 BuilderStep::Aggregate {
1223 config: camel_api::AggregatorConfig::correlate_by("id")
1224 .complete_when_size(1)
1225 .build()
1226 .unwrap(),
1227 },
1228 BuilderStep::Filter {
1229 predicate: FilterPredicate::new(|_: &Exchange| true),
1230 steps: vec![BuilderStep::Stop],
1231 },
1232 BuilderStep::WireTap {
1233 uri: "mock:tap".into(),
1234 },
1235 BuilderStep::DeclarativeLog {
1236 level: camel_processor::LogLevel::Info,
1237 message: ValueSourceDef::Expression(expr.clone()),
1238 },
1239 BuilderStep::Bean {
1240 name: "bean".into(),
1241 method: "call".into(),
1242 },
1243 BuilderStep::Script {
1244 language: "rhai".into(),
1245 script: "body".into(),
1246 },
1247 BuilderStep::Throttle {
1248 config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1249 steps: vec![BuilderStep::Stop],
1250 },
1251 BuilderStep::LoadBalance {
1252 config: camel_api::LoadBalancerConfig::round_robin(),
1253 steps: vec![BuilderStep::To("mock:l1".into())],
1254 },
1255 BuilderStep::DynamicRouter {
1256 config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1257 },
1258 BuilderStep::RoutingSlip {
1259 config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1260 },
1261 ];
1262
1263 for step in steps {
1264 let dbg = format!("{step:?}");
1265 assert!(!dbg.is_empty());
1266 }
1267 }
1268
1269 #[test]
1270 fn test_route_definition_to_info_preserves_metadata() {
1271 let info = RouteDefinition::new("direct:test", vec![])
1272 .with_route_id("meta-route")
1273 .with_auto_startup(false)
1274 .with_startup_order(7)
1275 .to_info();
1276
1277 assert_eq!(info.route_id(), "meta-route");
1278 assert!(!info.auto_startup());
1279 assert_eq!(info.startup_order(), 7);
1280 }
1281
1282 #[test]
1283 fn test_choice_builder_step_debug() {
1284 use camel_api::FilterPredicate;
1285
1286 fn always_true(_: &camel_api::Exchange) -> bool {
1287 true
1288 }
1289
1290 let step = BuilderStep::Choice {
1291 whens: vec![WhenStep {
1292 predicate: FilterPredicate::new(always_true),
1293 steps: vec![BuilderStep::To("mock:a".into())],
1294 }],
1295 otherwise: None,
1296 };
1297 let debug = format!("{step:?}");
1298 assert!(debug.contains("Choice"));
1299 }
1300
1301 #[test]
1302 fn test_route_definition_unit_of_work() {
1303 use camel_api::UnitOfWorkConfig;
1304 let config = UnitOfWorkConfig {
1305 on_complete: Some("log:complete".into()),
1306 on_failure: Some("log:failed".into()),
1307 };
1308 let def = RouteDefinition::new("direct:test", vec![])
1309 .with_route_id("uow-test")
1310 .with_unit_of_work(config.clone());
1311 assert_eq!(
1312 def.unit_of_work_config().unwrap().on_complete.as_deref(),
1313 Some("log:complete")
1314 );
1315 assert_eq!(
1316 def.unit_of_work_config().unwrap().on_failure.as_deref(),
1317 Some("log:failed")
1318 );
1319
1320 let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1321 assert!(def_no_uow.unit_of_work_config().is_none());
1322 }
1323
1324 #[test]
1325 fn test_route_definition_security_policy_accessor() {
1326 use async_trait::async_trait;
1327 use camel_api::CamelError;
1328 use camel_api::Exchange;
1329 use camel_api::security_policy::{
1330 AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1331 };
1332
1333 struct StubPolicy;
1334 #[async_trait]
1335 impl SecurityPolicy for StubPolicy {
1336 async fn evaluate(
1337 &self,
1338 _exchange: &mut Exchange,
1339 ) -> Result<AuthorizationDecision, CamelError> {
1340 Ok(AuthorizationDecision::Granted {
1341 principal: Principal {
1342 subject: "test".into(),
1343 issuer: "test".into(),
1344 audience: vec![],
1345 scopes: vec![],
1346 roles: vec![],
1347 claims: serde_json::Value::Null,
1348 },
1349 })
1350 }
1351 }
1352
1353 let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1354 assert!(def_no_sp.security_policy_config().is_none());
1355
1356 let def = RouteDefinition::new("direct:test", vec![])
1357 .with_route_id("sp-test")
1358 .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1359 assert!(def.security_policy_config().is_some());
1360 }
1361
1362 #[test]
1363 fn test_route_definition_security_authenticator_accessor() {
1364 use camel_api::security_policy::Principal;
1365
1366 struct TestAuth;
1367 #[async_trait::async_trait]
1368 impl TokenAuthenticator for TestAuth {
1369 async fn authenticate_bearer(
1370 &self,
1371 _token: &str,
1372 ) -> Result<Principal, camel_api::CamelError> {
1373 Ok(Principal {
1374 subject: "test".into(),
1375 issuer: "test".into(),
1376 audience: vec![],
1377 scopes: vec![],
1378 roles: vec![],
1379 claims: serde_json::Value::Null,
1380 })
1381 }
1382 }
1383
1384 let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1385 assert!(def_no_auth.security_authenticator().is_none());
1386
1387 let auth = Arc::new(TestAuth);
1388 let def = RouteDefinition::new("direct:test".to_string(), vec![])
1389 .with_security_authenticator(auth);
1390 assert!(def.security_authenticator().is_some());
1391 }
1392
1393 #[test]
1394 fn test_map_steps_swaps_steps_and_preserves_other_fields() {
1395 let original = RouteDefinition::new(
1396 "direct:test".to_string(),
1397 vec![BuilderStep::To("mock:a".into()), BuilderStep::Stop],
1398 )
1399 .with_route_id("my-route");
1400
1401 let mapped = original.map_steps(|steps| {
1402 let mut out = Vec::with_capacity(steps.len() + 1);
1403 out.push(BuilderStep::To("mock:prefix".into()));
1404 out.extend(steps);
1405 out
1406 });
1407
1408 assert_eq!(mapped.steps().len(), 3);
1410 assert!(matches!(mapped.steps()[0], BuilderStep::To(ref s) if s == "mock:prefix"));
1411 assert!(matches!(mapped.steps()[1], BuilderStep::To(ref s) if s == "mock:a"));
1412 assert_eq!(mapped.route_id(), "my-route");
1414 }
1415}