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) provider_registry: Option<Arc<camel_auth::ProviderRegistry>>,
351 pub(crate) security_provider: Option<String>,
355 pub(crate) security_audiences: Option<Vec<String>>,
360 pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
362 pub(crate) concurrency: Option<ConcurrencyModel>,
365 pub(crate) route_id: String,
367 pub(crate) auto_startup: bool,
369 pub(crate) startup_order: i32,
371 pub(crate) source_hash: Option<u64>,
372}
373
374impl RouteDefinition {
375 pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
377 Self {
378 from_uri: from_uri.into(),
379 steps,
380 error_handler: None,
381 circuit_breaker: None,
382 circuit_breaker_fallback: Vec::new(),
383 security_policy: None,
384 security_authenticator: None,
385 provider_registry: None,
386 security_provider: None,
387 security_audiences: None,
388 unit_of_work: None,
389 concurrency: None,
390 route_id: String::new(), auto_startup: true,
392 startup_order: 1000,
393 source_hash: None,
394 }
395 }
396
397 pub fn from_uri(&self) -> &str {
399 &self.from_uri
400 }
401
402 pub fn steps(&self) -> &[BuilderStep] {
404 &self.steps
405 }
406
407 pub fn map_steps(mut self, f: impl FnOnce(Vec<BuilderStep>) -> Vec<BuilderStep>) -> Self {
412 self.steps = f(self.steps);
413 self
414 }
415
416 pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
418 self.error_handler = Some(config);
419 self
420 }
421
422 pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
424 self.error_handler.as_ref()
425 }
426
427 pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
429 self.circuit_breaker = Some(config);
430 self
431 }
432
433 pub fn with_circuit_breaker_fallback(mut self, steps: Vec<BuilderStep>) -> Self {
440 self.circuit_breaker_fallback = steps;
441 self
442 }
443
444 pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
446 self.security_policy = Some(config);
447 self
448 }
449
450 pub fn with_security_authenticator(
452 mut self,
453 authenticator: Arc<dyn TokenAuthenticator>,
454 ) -> Self {
455 self.security_authenticator = Some(authenticator);
456 self
457 }
458
459 pub fn with_provider_registry(mut self, registry: Arc<camel_auth::ProviderRegistry>) -> Self {
465 self.provider_registry = Some(registry);
466 self
467 }
468
469 pub fn with_security_provider(mut self, name: impl Into<String>) -> Self {
471 self.security_provider = Some(name.into());
472 self
473 }
474
475 pub fn with_security_audiences(mut self, audiences: Vec<String>) -> Self {
477 self.security_audiences = Some(audiences);
478 self
479 }
480
481 pub fn security_provider(&self) -> Option<&str> {
483 self.security_provider.as_deref()
484 }
485
486 pub fn security_audiences(&self) -> Option<&[String]> {
488 self.security_audiences.as_deref()
489 }
490
491 pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
493 self.unit_of_work = Some(config);
494 self
495 }
496
497 pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
499 self.unit_of_work.as_ref()
500 }
501
502 pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
504 self.circuit_breaker.as_ref()
505 }
506
507 pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
508 self.security_policy.as_ref()
509 }
510
511 pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
512 self.security_authenticator.as_ref()
513 }
514
515 pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
517 self.concurrency.as_ref()
518 }
519
520 pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
522 self.concurrency = Some(model);
523 self
524 }
525
526 pub fn route_id(&self) -> &str {
528 &self.route_id
529 }
530
531 pub fn auto_startup(&self) -> bool {
533 self.auto_startup
534 }
535
536 pub fn startup_order(&self) -> i32 {
538 self.startup_order
539 }
540
541 pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
543 self.route_id = id.into();
544 self
545 }
546
547 pub fn with_auto_startup(mut self, auto: bool) -> Self {
549 self.auto_startup = auto;
550 self
551 }
552
553 pub fn with_startup_order(mut self, order: i32) -> Self {
555 self.startup_order = order;
556 self
557 }
558
559 pub fn with_source_hash(mut self, hash: u64) -> Self {
560 self.source_hash = Some(hash);
561 self
562 }
563
564 pub fn source_hash(&self) -> Option<u64> {
565 self.source_hash
566 }
567
568 pub fn to_info(&self) -> RouteDefinitionInfo {
571 RouteDefinitionInfo {
572 route_id: self.route_id.clone(),
573 auto_startup: self.auto_startup,
574 startup_order: self.startup_order,
575 source_hash: self.source_hash,
576 }
577 }
578}
579
580#[derive(Clone)]
586pub struct RouteDefinitionInfo {
587 route_id: String,
588 auto_startup: bool,
589 startup_order: i32,
590 pub(crate) source_hash: Option<u64>,
591}
592
593impl RouteDefinitionInfo {
594 pub fn route_id(&self) -> &str {
596 &self.route_id
597 }
598
599 pub fn auto_startup(&self) -> bool {
601 self.auto_startup
602 }
603
604 pub fn startup_order(&self) -> i32 {
606 self.startup_order
607 }
608
609 pub fn source_hash(&self) -> Option<u64> {
610 self.source_hash
611 }
612}
613
614#[cfg(test)]
615mod tests {
616 use super::*;
617
618 #[test]
623 fn golden_debug_output_all_variants() {
624 use camel_api::declarative::LanguageExpressionDef;
625 use camel_api::loop_eip::LoopMode;
626 use camel_api::recipient_list::RecipientListConfig;
627 use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
628 use camel_api::{
629 BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
630 FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
631 Value,
632 };
633 use std::sync::Arc;
634
635 let expr = LanguageExpressionDef {
636 language: "simple".into(),
637 source: "${body}".into(),
638 };
639
640 assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
643 assert_eq!(
644 format!(
645 "{:?}",
646 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
647 ),
648 "Processor(BoxProcessor(...))"
649 );
650 assert_eq!(
651 format!("{:?}", BuilderStep::To("mock:out".into())),
652 "To(\"mock:out\")"
653 );
654
655 assert_eq!(
658 format!(
659 "{:?}",
660 BuilderStep::Log {
661 level: camel_processor::LogLevel::Info,
662 message: "hello".into(),
663 }
664 ),
665 "Log { level: Info, message: \"hello\" }"
666 );
667
668 assert_eq!(
669 format!(
670 "{:?}",
671 BuilderStep::DeclarativeSetHeader {
672 key: "k".into(),
673 value: ValueSourceDef::Literal(Value::String("v".into())),
674 }
675 ),
676 "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
677 );
678
679 assert_eq!(
680 format!(
681 "{:?}",
682 BuilderStep::DeclarativeSetHeaderIfAbsent {
683 key: "k".into(),
684 value: ValueSourceDef::Literal(Value::String("v".into())),
685 }
686 ),
687 "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
688 );
689
690 assert_eq!(
691 format!(
692 "{:?}",
693 BuilderStep::DeclarativeSetBody {
694 value: ValueSourceDef::Literal(Value::String("v".into())),
695 }
696 ),
697 "DeclarativeSetBody { value: Literal(String(\"v\")) }"
698 );
699
700 assert_eq!(
701 format!(
702 "{:?}",
703 BuilderStep::DeclarativeSetProperty {
704 key: "prop".into(),
705 value_source: ValueSourceDef::Literal(Value::String("v".into())),
706 }
707 ),
708 "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
709 );
710
711 assert_eq!(
712 format!(
713 "{:?}",
714 BuilderStep::DeclarativeScript {
715 expression: expr.clone(),
716 }
717 ),
718 "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
719 );
720
721 let func_def = FunctionDefinition {
723 id: FunctionId("test-id".into()),
724 runtime: "my_runtime".into(),
725 source: "${body}".into(),
726 timeout_ms: 5000,
727 route_id: None,
728 step_index: None,
729 };
730 assert_eq!(
731 format!(
732 "{:?}",
733 BuilderStep::DeclarativeFunction {
734 definition: func_def,
735 }
736 ),
737 "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
738 );
739
740 assert_eq!(
741 format!(
742 "{:?}",
743 BuilderStep::WireTap {
744 uri: "mock:tap".into(),
745 }
746 ),
747 "WireTap { uri: \"mock:tap\" }"
748 );
749
750 assert_eq!(
751 format!(
752 "{:?}",
753 BuilderStep::DeclarativeLog {
754 level: camel_processor::LogLevel::Info,
755 message: ValueSourceDef::Expression(expr.clone()),
756 }
757 ),
758 "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
759 );
760
761 assert_eq!(
762 format!(
763 "{:?}",
764 BuilderStep::Bean {
765 name: "myBean".into(),
766 method: "process".into(),
767 }
768 ),
769 "Bean { name: \"myBean\", method: \"process\" }"
770 );
771
772 assert_eq!(
773 format!(
774 "{:?}",
775 BuilderStep::Script {
776 language: "js".into(),
777 script: "body".into(),
778 }
779 ),
780 "Script { language: \"js\", script: \"body\" }"
781 );
782
783 assert_eq!(
784 format!(
785 "{:?}",
786 BuilderStep::Aggregate {
787 config: camel_api::AggregatorConfig::correlate_by("id")
788 .complete_when_size(1)
789 .build()
790 .unwrap(),
791 }
792 ),
793 "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 } }"
794 );
795
796 assert_eq!(
797 format!(
798 "{:?}",
799 BuilderStep::DynamicRouter {
800 config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
801 "mock:dr".into()
802 ))),
803 }
804 ),
805 "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
806 );
807
808 assert_eq!(
809 format!(
810 "{:?}",
811 BuilderStep::RoutingSlip {
812 config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
813 }
814 ),
815 "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
816 );
817
818 assert_eq!(
819 format!(
820 "{:?}",
821 BuilderStep::RecipientList {
822 config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
823 }
824 ),
825 "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
826 );
827
828 assert_eq!(
829 format!(
830 "{:?}",
831 BuilderStep::Enrich {
832 uri: "mock:enrich".into(),
833 strategy: Some("agg".into()),
834 timeout_ms: Some(1000),
835 }
836 ),
837 "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
838 );
839
840 assert_eq!(
841 format!(
842 "{:?}",
843 BuilderStep::PollEnrich {
844 uri: "mock:poll".into(),
845 strategy: None,
846 timeout_ms: None,
847 }
848 ),
849 "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
850 );
851
852 assert_eq!(
853 format!(
854 "{:?}",
855 BuilderStep::Validate {
856 predicate: expr.clone(),
857 }
858 ),
859 "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
860 );
861
862 assert_eq!(
863 format!("{:?}", BuilderStep::Sampling { period: 100 }),
864 "Sampling { period: 100 }"
865 );
866
867 assert_eq!(
868 format!(
869 "{:?}",
870 BuilderStep::Resequence {
871 policy_config: Default::default(),
872 }
873 ),
874 "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
875 );
876
877 assert_eq!(
879 format!(
880 "{:?}",
881 BuilderStep::DeclarativeFilter {
882 predicate: expr.clone(),
883 steps: vec![BuilderStep::Stop],
884 }
885 ),
886 "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
887 );
888
889 assert_eq!(
890 format!(
891 "{:?}",
892 BuilderStep::DeclarativeSplit {
893 expression: expr.clone(),
894 aggregation: AggregationStrategy::Original,
895 parallel: false,
896 parallel_limit: Some(2),
897 stop_on_exception: true,
898 steps: vec![BuilderStep::Stop],
899 }
900 ),
901 "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
902 );
903
904 assert_eq!(
905 format!(
906 "{:?}",
907 BuilderStep::Split {
908 config: camel_api::splitter::SplitterConfig::new(
909 camel_api::splitter::split_body_lines()
910 ),
911 steps: vec![BuilderStep::Stop],
912 }
913 ),
914 "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
915 );
916
917 assert_eq!(
918 format!(
919 "{:?}",
920 BuilderStep::Filter {
921 predicate: FilterPredicate::new(|_: &Exchange| true),
922 steps: vec![BuilderStep::Stop],
923 }
924 ),
925 "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
926 );
927
928 assert_eq!(
929 format!(
930 "{:?}",
931 BuilderStep::Throttle {
932 config: camel_api::ThrottlerConfig::new(
933 10,
934 std::time::Duration::from_millis(10)
935 ),
936 steps: vec![BuilderStep::Stop],
937 }
938 ),
939 "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
940 );
941
942 assert_eq!(
943 format!(
944 "{:?}",
945 BuilderStep::LoadBalance {
946 config: camel_api::LoadBalancerConfig::round_robin(),
947 steps: vec![BuilderStep::To("mock:l1".into())],
948 }
949 ),
950 "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
951 );
952
953 assert_eq!(
954 format!(
955 "{:?}",
956 BuilderStep::Delay {
957 config: camel_api::DelayConfig::new(500),
958 }
959 ),
960 "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
961 );
962
963 assert_eq!(
967 format!(
968 "{:?}",
969 BuilderStep::Choice {
970 whens: vec![WhenStep {
971 predicate: FilterPredicate::new(|_: &Exchange| true),
972 steps: vec![BuilderStep::To("mock:a".into())],
973 }],
974 otherwise: None,
975 }
976 ),
977 "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
978 );
979
980 assert_eq!(
981 format!(
982 "{:?}",
983 BuilderStep::DeclarativeChoice {
984 whens: vec![DeclarativeWhenStep {
985 predicate: expr.clone(),
986 steps: vec![BuilderStep::Stop],
987 }],
988 otherwise: Some(vec![BuilderStep::Stop]),
989 }
990 ),
991 "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
992 );
993
994 assert_eq!(
996 format!(
997 "{:?}",
998 BuilderStep::Multicast {
999 steps: vec![BuilderStep::To("direct:a".into())],
1000 config: MulticastConfig::new(),
1001 }
1002 ),
1003 "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
1004 );
1005
1006 assert_eq!(
1008 format!(
1009 "{:?}",
1010 BuilderStep::DeclarativeDynamicRouter {
1011 expression: expr.clone(),
1012 uri_delimiter: ",".into(),
1013 cache_size: 1000,
1014 ignore_invalid_endpoints: false,
1015 max_iterations: 1000,
1016 }
1017 ),
1018 "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
1019 );
1020
1021 assert_eq!(
1022 format!(
1023 "{:?}",
1024 BuilderStep::DeclarativeRoutingSlip {
1025 expression: expr.clone(),
1026 uri_delimiter: ",".into(),
1027 cache_size: 1000,
1028 ignore_invalid_endpoints: false,
1029 }
1030 ),
1031 "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
1032 );
1033
1034 assert_eq!(
1036 format!(
1037 "{:?}",
1038 BuilderStep::DeclarativeRecipientList {
1039 expression: expr.clone(),
1040 delimiter: ",".into(),
1041 parallel: false,
1042 parallel_limit: None,
1043 stop_on_exception: false,
1044 aggregation: "original".into(),
1045 }
1046 ),
1047 "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
1048 );
1049
1050 assert_eq!(
1052 format!(
1053 "{:?}",
1054 BuilderStep::Loop {
1055 config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
1056 steps: vec![],
1057 }
1058 ),
1059 "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
1060 );
1061
1062 assert_eq!(
1063 format!(
1064 "{:?}",
1065 BuilderStep::DeclarativeLoop {
1066 count: Some(5),
1067 while_predicate: None,
1068 steps: vec![],
1069 max_iterations: Some(100),
1070 }
1071 ),
1072 "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
1073 );
1074
1075 assert_eq!(
1077 format!(
1078 "{:?}",
1079 BuilderStep::ClaimCheck {
1080 repository: "myRepo".into(),
1081 operation: "checkout".into(),
1082 key: expr.clone(),
1083 filter: None,
1084 }
1085 ),
1086 "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
1087 );
1088
1089 assert_eq!(
1091 format!(
1092 "{:?}",
1093 BuilderStep::Sort {
1094 expression: expr.clone(),
1095 reverse: false,
1096 }
1097 ),
1098 "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
1099 );
1100
1101 assert_eq!(
1103 format!(
1104 "{:?}",
1105 BuilderStep::IdempotentConsumer {
1106 repository: "myRepo".into(),
1107 expression: expr.clone(),
1108 steps: vec![],
1109 eager: true,
1110 remove_on_failure: false,
1111 }
1112 ),
1113 "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
1114 );
1115
1116 assert_eq!(
1118 format!(
1119 "{:?}",
1120 BuilderStep::DeclarativeDoTry {
1121 try_steps: vec![BuilderStep::Stop],
1122 catch: vec![],
1123 finally: None,
1124 }
1125 ),
1126 "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1127 );
1128
1129 assert_eq!(
1131 format!(
1132 "{:?}",
1133 BuilderStep::DeclarativeStreamSplit {
1134 stream_config: StreamSplitConfig {
1135 format: StreamSplitFormat::Ndjson,
1136 max_record_bytes: 1024 * 1024,
1137 batch_size: 1,
1138 chunk_size: None,
1139 include_origin: true,
1140 },
1141 aggregation: AggregationStrategy::Original,
1142 stop_on_exception: true,
1143 steps: vec![BuilderStep::Stop],
1144 }
1145 ),
1146 "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] }"
1147 );
1148 }
1149
1150 #[test]
1151 fn test_builder_step_multicast_variant() {
1152 use camel_api::MulticastConfig;
1153
1154 let step = BuilderStep::Multicast {
1155 steps: vec![BuilderStep::To("direct:a".into())],
1156 config: MulticastConfig::new(),
1157 };
1158
1159 assert!(matches!(step, BuilderStep::Multicast { .. }));
1160 }
1161
1162 #[test]
1163 fn test_route_definition_defaults() {
1164 let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1165 assert_eq!(def.route_id(), "test-route");
1166 assert!(def.auto_startup());
1167 assert_eq!(def.startup_order(), 1000);
1168 }
1169
1170 #[test]
1171 fn test_route_definition_builders() {
1172 let def = RouteDefinition::new("direct:test", vec![])
1173 .with_route_id("my-route")
1174 .with_auto_startup(false)
1175 .with_startup_order(50);
1176 assert_eq!(def.route_id(), "my-route");
1177 assert!(!def.auto_startup());
1178 assert_eq!(def.startup_order(), 50);
1179 }
1180
1181 #[test]
1182 fn test_route_definition_accessors_cover_core_fields() {
1183 let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1184 .with_route_id("accessor-route");
1185
1186 assert_eq!(def.from_uri(), "direct:in");
1187 assert_eq!(def.steps().len(), 1);
1188 assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1189 }
1190
1191 #[test]
1192 fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1193 use camel_api::circuit_breaker::CircuitBreakerConfig;
1194 use camel_api::error_handler::ErrorHandlerConfig;
1195 use camel_component_api::ConcurrencyModel;
1196
1197 let def = RouteDefinition::new("direct:test", vec![])
1198 .with_route_id("eh-route")
1199 .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1200 .with_circuit_breaker(CircuitBreakerConfig::new())
1201 .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1202
1203 let eh = def
1204 .error_handler_config()
1205 .expect("error handler should be set");
1206 assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1207 assert!(def.circuit_breaker_config().is_some());
1208 assert!(matches!(
1209 def.concurrency_override(),
1210 Some(ConcurrencyModel::Concurrent { max: Some(4) })
1211 ));
1212 }
1213
1214 #[test]
1215 fn test_builder_step_debug_covers_many_variants() {
1216 use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1217 use camel_api::{
1218 BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1219 OpaqueProcessor, RoutingSlipConfig, Value,
1220 };
1221 use std::sync::Arc;
1222
1223 let expr = LanguageExpressionDef {
1224 language: "simple".into(),
1225 source: "${body}".into(),
1226 };
1227
1228 let steps = vec![
1229 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1230 BuilderStep::To("mock:out".into()),
1231 BuilderStep::Stop,
1232 BuilderStep::Log {
1233 level: camel_processor::LogLevel::Info,
1234 message: "hello".into(),
1235 },
1236 BuilderStep::DeclarativeSetHeader {
1237 key: "k".into(),
1238 value: ValueSourceDef::Literal(Value::String("v".into())),
1239 },
1240 BuilderStep::DeclarativeSetBody {
1241 value: ValueSourceDef::Expression(expr.clone()),
1242 },
1243 BuilderStep::DeclarativeFilter {
1244 predicate: expr.clone(),
1245 steps: vec![BuilderStep::Stop],
1246 },
1247 BuilderStep::DeclarativeChoice {
1248 whens: vec![DeclarativeWhenStep {
1249 predicate: expr.clone(),
1250 steps: vec![BuilderStep::Stop],
1251 }],
1252 otherwise: Some(vec![BuilderStep::Stop]),
1253 },
1254 BuilderStep::DeclarativeScript {
1255 expression: expr.clone(),
1256 },
1257 BuilderStep::DeclarativeSplit {
1258 expression: expr.clone(),
1259 aggregation: AggregationStrategy::Original,
1260 parallel: false,
1261 parallel_limit: Some(2),
1262 stop_on_exception: true,
1263 steps: vec![BuilderStep::Stop],
1264 },
1265 BuilderStep::Split {
1266 config: SplitterConfig::new(split_body_lines()),
1267 steps: vec![BuilderStep::Stop],
1268 },
1269 BuilderStep::Aggregate {
1270 config: camel_api::AggregatorConfig::correlate_by("id")
1271 .complete_when_size(1)
1272 .build()
1273 .unwrap(),
1274 },
1275 BuilderStep::Filter {
1276 predicate: FilterPredicate::new(|_: &Exchange| true),
1277 steps: vec![BuilderStep::Stop],
1278 },
1279 BuilderStep::WireTap {
1280 uri: "mock:tap".into(),
1281 },
1282 BuilderStep::DeclarativeLog {
1283 level: camel_processor::LogLevel::Info,
1284 message: ValueSourceDef::Expression(expr.clone()),
1285 },
1286 BuilderStep::Bean {
1287 name: "bean".into(),
1288 method: "call".into(),
1289 },
1290 BuilderStep::Script {
1291 language: "rhai".into(),
1292 script: "body".into(),
1293 },
1294 BuilderStep::Throttle {
1295 config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1296 steps: vec![BuilderStep::Stop],
1297 },
1298 BuilderStep::LoadBalance {
1299 config: camel_api::LoadBalancerConfig::round_robin(),
1300 steps: vec![BuilderStep::To("mock:l1".into())],
1301 },
1302 BuilderStep::DynamicRouter {
1303 config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1304 },
1305 BuilderStep::RoutingSlip {
1306 config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1307 },
1308 ];
1309
1310 for step in steps {
1311 let dbg = format!("{step:?}");
1312 assert!(!dbg.is_empty());
1313 }
1314 }
1315
1316 #[test]
1317 fn test_route_definition_to_info_preserves_metadata() {
1318 let info = RouteDefinition::new("direct:test", vec![])
1319 .with_route_id("meta-route")
1320 .with_auto_startup(false)
1321 .with_startup_order(7)
1322 .to_info();
1323
1324 assert_eq!(info.route_id(), "meta-route");
1325 assert!(!info.auto_startup());
1326 assert_eq!(info.startup_order(), 7);
1327 }
1328
1329 #[test]
1330 fn test_choice_builder_step_debug() {
1331 use camel_api::FilterPredicate;
1332
1333 fn always_true(_: &camel_api::Exchange) -> bool {
1334 true
1335 }
1336
1337 let step = BuilderStep::Choice {
1338 whens: vec![WhenStep {
1339 predicate: FilterPredicate::new(always_true),
1340 steps: vec![BuilderStep::To("mock:a".into())],
1341 }],
1342 otherwise: None,
1343 };
1344 let debug = format!("{step:?}");
1345 assert!(debug.contains("Choice"));
1346 }
1347
1348 #[test]
1349 fn test_route_definition_unit_of_work() {
1350 use camel_api::UnitOfWorkConfig;
1351 let config = UnitOfWorkConfig {
1352 on_complete: Some("log:complete".into()),
1353 on_failure: Some("log:failed".into()),
1354 };
1355 let def = RouteDefinition::new("direct:test", vec![])
1356 .with_route_id("uow-test")
1357 .with_unit_of_work(config.clone());
1358 assert_eq!(
1359 def.unit_of_work_config().unwrap().on_complete.as_deref(),
1360 Some("log:complete")
1361 );
1362 assert_eq!(
1363 def.unit_of_work_config().unwrap().on_failure.as_deref(),
1364 Some("log:failed")
1365 );
1366
1367 let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1368 assert!(def_no_uow.unit_of_work_config().is_none());
1369 }
1370
1371 #[test]
1372 fn test_route_definition_security_policy_accessor() {
1373 use async_trait::async_trait;
1374 use camel_api::CamelError;
1375 use camel_api::Exchange;
1376 use camel_api::security_policy::{
1377 AuthContext, AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1378 };
1379
1380 struct StubPolicy;
1381 #[async_trait]
1382 impl SecurityPolicy for StubPolicy {
1383 async fn evaluate(
1384 &self,
1385 _exchange: &mut Exchange,
1386 _auth: &AuthContext<'_>,
1387 ) -> Result<AuthorizationDecision, CamelError> {
1388 Ok(AuthorizationDecision::Granted {
1389 principal: Principal {
1390 subject: "test".into(),
1391 issuer: "test".into(),
1392 audience: vec![],
1393 scopes: vec![],
1394 roles: vec![],
1395 claims: serde_json::Value::Null,
1396 },
1397 })
1398 }
1399 }
1400
1401 let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1402 assert!(def_no_sp.security_policy_config().is_none());
1403
1404 let def = RouteDefinition::new("direct:test", vec![])
1405 .with_route_id("sp-test")
1406 .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1407 assert!(def.security_policy_config().is_some());
1408 }
1409
1410 #[test]
1411 fn test_route_definition_security_authenticator_accessor() {
1412 use camel_api::security_policy::Principal;
1413
1414 struct TestAuth;
1415 #[async_trait::async_trait]
1416 impl TokenAuthenticator for TestAuth {
1417 async fn authenticate_bearer(
1418 &self,
1419 _token: &str,
1420 ) -> Result<Principal, camel_api::CamelError> {
1421 Ok(Principal {
1422 subject: "test".into(),
1423 issuer: "test".into(),
1424 audience: vec![],
1425 scopes: vec![],
1426 roles: vec![],
1427 claims: serde_json::Value::Null,
1428 })
1429 }
1430 }
1431
1432 let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1433 assert!(def_no_auth.security_authenticator().is_none());
1434
1435 let auth = Arc::new(TestAuth);
1436 let def = RouteDefinition::new("direct:test".to_string(), vec![])
1437 .with_security_authenticator(auth);
1438 assert!(def.security_authenticator().is_some());
1439 }
1440
1441 #[test]
1442 fn test_map_steps_swaps_steps_and_preserves_other_fields() {
1443 let original = RouteDefinition::new(
1444 "direct:test".to_string(),
1445 vec![BuilderStep::To("mock:a".into()), BuilderStep::Stop],
1446 )
1447 .with_route_id("my-route");
1448
1449 let mapped = original.map_steps(|steps| {
1450 let mut out = Vec::with_capacity(steps.len() + 1);
1451 out.push(BuilderStep::To("mock:prefix".into()));
1452 out.extend(steps);
1453 out
1454 });
1455
1456 assert_eq!(mapped.steps().len(), 3);
1458 assert!(matches!(mapped.steps()[0], BuilderStep::To(ref s) if s == "mock:prefix"));
1459 assert!(matches!(mapped.steps()[1], BuilderStep::To(ref s) if s == "mock:a"));
1460 assert_eq!(mapped.route_id(), "my-route");
1462 }
1463}