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 SpanKindHint, 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
329impl BuilderStep {
330 pub(crate) fn span_label(&self) -> Option<String> {
344 match self {
345 Self::Processor(_) | Self::Stop => None,
347
348 Self::To(uri) => uri.contains(':').then(|| {
349 let scheme = uri.split(':').next().unwrap_or_default();
350 format!("to:{scheme}")
351 }),
352
353 Self::Log { .. } | Self::DeclarativeLog { .. } => Some("log".into()),
354
355 Self::DeclarativeSetHeader { .. } => Some("set-header".into()),
356 Self::DeclarativeSetHeaderIfAbsent { .. } => Some("set-header-if-absent".into()),
357 Self::DeclarativeRemoveHeader { .. } => Some("remove-header".into()),
358 Self::DeclarativeSetProperty { .. } => Some("set-property".into()),
359 Self::DeclarativeSetBody { .. } => Some("set-body".into()),
360
361 Self::DeclarativeFilter { .. } | Self::Filter { .. } => Some("filter".into()),
362 Self::DeclarativeChoice { .. } | Self::Choice { .. } => Some("choice".into()),
363 Self::DeclarativeScript { .. } | Self::Script { .. } => Some("script".into()),
364 Self::DeclarativeFunction { .. } => Some("function".into()),
365
366 Self::DeclarativeSplit { .. }
367 | Self::DeclarativeStreamSplit { .. }
368 | Self::Split { .. } => Some("split".into()),
369
370 Self::DeclarativeDynamicRouter { .. } | Self::DynamicRouter { .. } => {
371 Some("dynamic-router".into())
372 }
373 Self::DeclarativeRoutingSlip { .. } | Self::RoutingSlip { .. } => {
374 Some("routing-slip".into())
375 }
376 Self::DeclarativeRecipientList { .. } | Self::RecipientList { .. } => {
377 Some("recipient-list".into())
378 }
379
380 Self::Aggregate { .. } => Some("aggregate".into()),
381 Self::WireTap { .. } => Some("wire-tap".into()),
382 Self::Multicast { .. } => Some("multicast".into()),
383 Self::Bean { .. } => Some("bean".into()),
384 Self::Throttle { .. } => Some("throttle".into()),
385 Self::LoadBalance { .. } => Some("load-balance".into()),
386 Self::Delay { .. } => Some("delay".into()),
387 Self::Loop { .. } | Self::DeclarativeLoop { .. } => Some("loop".into()),
388 Self::Enrich { .. } => Some("enrich".into()),
389 Self::PollEnrich { .. } => Some("poll-enrich".into()),
390 Self::Validate { .. } => Some("validate".into()),
391 Self::ClaimCheck { .. } => Some("claim-check".into()),
392 Self::Sampling { .. } => Some("sampling".into()),
393 Self::Sort { .. } => Some("sort".into()),
394 Self::IdempotentConsumer { .. } => Some("idempotent-consumer".into()),
395 Self::Cache { .. } => Some("cache".into()),
396 Self::CacheInvalidate { .. } => Some("cache-invalidate".into()),
397 Self::CacheClear { .. } => Some("cache-clear".into()),
398 Self::CacheStats { .. } => Some("cache-stats".into()),
399 Self::CachePeekStale { .. } => Some("cache-peek-stale".into()),
400 Self::DeclarativeDoTry { .. } => Some("do-try".into()),
401 Self::Resequence { .. } => Some("resequence".into()),
402 }
403 }
404
405 pub(crate) fn to_uri_metadata(&self) -> Option<Arc<str>> {
416 match self {
417 Self::To(uri) => Some(Arc::from(uri.as_str())),
418
419 Self::Processor(..)
423 | Self::Stop
424 | Self::Log { .. }
425 | Self::DeclarativeSetHeader { .. }
426 | Self::DeclarativeSetHeaderIfAbsent { .. }
427 | Self::DeclarativeRemoveHeader { .. }
428 | Self::DeclarativeSetProperty { .. }
429 | Self::DeclarativeSetBody { .. }
430 | Self::DeclarativeFilter { .. }
431 | Self::DeclarativeChoice { .. }
432 | Self::DeclarativeScript { .. }
433 | Self::DeclarativeFunction { .. }
434 | Self::DeclarativeSplit { .. }
435 | Self::DeclarativeStreamSplit { .. }
436 | Self::DeclarativeDynamicRouter { .. }
437 | Self::DeclarativeRoutingSlip { .. }
438 | Self::Split { .. }
439 | Self::Aggregate { .. }
440 | Self::Filter { .. }
441 | Self::Choice { .. }
442 | Self::Multicast { .. }
443 | Self::DeclarativeLog { .. }
444 | Self::Bean { .. }
445 | Self::Script { .. }
446 | Self::Throttle { .. }
447 | Self::LoadBalance { .. }
448 | Self::DynamicRouter { .. }
449 | Self::RoutingSlip { .. }
450 | Self::RecipientList { .. }
451 | Self::DeclarativeRecipientList { .. }
452 | Self::Delay { .. }
453 | Self::Loop { .. }
454 | Self::DeclarativeLoop { .. }
455 | Self::Enrich { .. }
456 | Self::PollEnrich { .. }
457 | Self::WireTap { .. }
458 | Self::Validate { .. }
459 | Self::ClaimCheck { .. }
460 | Self::Sampling { .. }
461 | Self::Sort { .. }
462 | Self::IdempotentConsumer { .. }
463 | Self::Cache { .. }
464 | Self::CacheInvalidate { .. }
465 | Self::CacheClear { .. }
466 | Self::CacheStats { .. }
467 | Self::CachePeekStale { .. }
468 | Self::DeclarativeDoTry { .. }
469 | Self::Resequence { .. } => None,
470 }
471 }
472
473 pub(crate) fn span_kind_hint(&self) -> SpanKindHint {
489 match self {
490 Self::To(uri)
492 | Self::Enrich { uri, .. }
493 | Self::PollEnrich { uri, .. }
494 | Self::WireTap { uri, .. } => uri_span_kind(uri),
495
496 Self::Processor(..)
500 | Self::Stop
501 | Self::Log { .. }
502 | Self::DeclarativeSetHeader { .. }
503 | Self::DeclarativeSetHeaderIfAbsent { .. }
504 | Self::DeclarativeRemoveHeader { .. }
505 | Self::DeclarativeSetProperty { .. }
506 | Self::DeclarativeSetBody { .. }
507 | Self::DeclarativeFilter { .. }
508 | Self::DeclarativeChoice { .. }
509 | Self::DeclarativeScript { .. }
510 | Self::DeclarativeFunction { .. }
511 | Self::DeclarativeSplit { .. }
512 | Self::DeclarativeStreamSplit { .. }
513 | Self::DeclarativeDynamicRouter { .. }
514 | Self::DeclarativeRoutingSlip { .. }
515 | Self::Split { .. }
516 | Self::Aggregate { .. }
517 | Self::Filter { .. }
518 | Self::Choice { .. }
519 | Self::Multicast { .. }
520 | Self::DeclarativeLog { .. }
521 | Self::Bean { .. }
522 | Self::Script { .. }
523 | Self::Throttle { .. }
524 | Self::LoadBalance { .. }
525 | Self::DynamicRouter { .. }
526 | Self::RoutingSlip { .. }
527 | Self::RecipientList { .. }
528 | Self::DeclarativeRecipientList { .. }
529 | Self::Delay { .. }
530 | Self::Loop { .. }
531 | Self::DeclarativeLoop { .. }
532 | Self::Validate { .. }
533 | Self::ClaimCheck { .. }
534 | Self::Sampling { .. }
535 | Self::Sort { .. }
536 | Self::IdempotentConsumer { .. }
537 | Self::Cache { .. }
538 | Self::CacheInvalidate { .. }
539 | Self::CacheClear { .. }
540 | Self::CacheStats { .. }
541 | Self::CachePeekStale { .. }
542 | Self::DeclarativeDoTry { .. }
543 | Self::Resequence { .. } => SpanKindHint::Internal,
544 }
545 }
546}
547
548fn uri_span_kind(uri: &str) -> SpanKindHint {
555 const PRODUCER_SCHEMES: [&str; 5] = ["kafka", "jms", "activemq", "artemis", "mqtt"];
557 const CLIENT_SCHEMES: [&str; 12] = [
559 "http",
560 "https",
561 "grpc",
562 "grpcs",
563 "ws",
564 "redis",
565 "opensearch",
566 "sql",
567 "surrealdb",
568 "cxf",
569 "llm",
570 "mcp",
571 ];
572
573 let scheme = uri.split(':').next().unwrap_or_default();
574 if PRODUCER_SCHEMES
575 .iter()
576 .any(|s| s.eq_ignore_ascii_case(scheme))
577 {
578 SpanKindHint::Producer
579 } else if CLIENT_SCHEMES
580 .iter()
581 .any(|s| s.eq_ignore_ascii_case(scheme))
582 {
583 SpanKindHint::Client
584 } else {
585 SpanKindHint::Internal
586 }
587}
588
589pub struct RouteDefinition {
591 pub(crate) from_uri: String,
592 pub(crate) steps: Vec<BuilderStep>,
593 pub(crate) error_handler: Option<ErrorHandlerConfig>,
595 pub(crate) circuit_breaker: Option<CircuitBreakerConfig>,
597 pub(crate) circuit_breaker_fallback: Vec<BuilderStep>,
605 pub(crate) security_policy: Option<SecurityPolicyConfig>,
606 pub(crate) security_authenticator: Option<Arc<dyn TokenAuthenticator>>,
608 pub(crate) provider_registry: Option<Arc<camel_auth::ProviderRegistry>>,
611 pub(crate) security_provider: Option<String>,
615 pub(crate) security_audiences: Option<Vec<String>>,
620 pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
622 pub(crate) concurrency: Option<ConcurrencyModel>,
625 pub(crate) route_id: String,
627 pub(crate) auto_startup: bool,
629 pub(crate) startup_order: i32,
631 pub(crate) source_hash: Option<u64>,
632}
633
634impl RouteDefinition {
635 pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
637 Self {
638 from_uri: from_uri.into(),
639 steps,
640 error_handler: None,
641 circuit_breaker: None,
642 circuit_breaker_fallback: Vec::new(),
643 security_policy: None,
644 security_authenticator: None,
645 provider_registry: None,
646 security_provider: None,
647 security_audiences: None,
648 unit_of_work: None,
649 concurrency: None,
650 route_id: String::new(), auto_startup: true,
652 startup_order: 1000,
653 source_hash: None,
654 }
655 }
656
657 pub fn from_uri(&self) -> &str {
659 &self.from_uri
660 }
661
662 pub fn steps(&self) -> &[BuilderStep] {
664 &self.steps
665 }
666
667 pub fn circuit_breaker_fallback(&self) -> &[BuilderStep] {
668 &self.circuit_breaker_fallback
669 }
670
671 pub fn map_steps(mut self, f: impl FnOnce(Vec<BuilderStep>) -> Vec<BuilderStep>) -> Self {
676 self.steps = f(self.steps);
677 self
678 }
679
680 pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
682 self.error_handler = Some(config);
683 self
684 }
685
686 pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
688 self.error_handler.as_ref()
689 }
690
691 pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
693 self.circuit_breaker = Some(config);
694 self
695 }
696
697 pub fn with_circuit_breaker_fallback(mut self, steps: Vec<BuilderStep>) -> Self {
704 self.circuit_breaker_fallback = steps;
705 self
706 }
707
708 pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
710 self.security_policy = Some(config);
711 self
712 }
713
714 pub fn with_security_authenticator(
716 mut self,
717 authenticator: Arc<dyn TokenAuthenticator>,
718 ) -> Self {
719 self.security_authenticator = Some(authenticator);
720 self
721 }
722
723 pub fn with_provider_registry(mut self, registry: Arc<camel_auth::ProviderRegistry>) -> Self {
729 self.provider_registry = Some(registry);
730 self
731 }
732
733 pub fn with_security_provider(mut self, name: impl Into<String>) -> Self {
735 self.security_provider = Some(name.into());
736 self
737 }
738
739 pub fn with_security_audiences(mut self, audiences: Vec<String>) -> Self {
741 self.security_audiences = Some(audiences);
742 self
743 }
744
745 pub fn security_provider(&self) -> Option<&str> {
747 self.security_provider.as_deref()
748 }
749
750 pub fn security_audiences(&self) -> Option<&[String]> {
752 self.security_audiences.as_deref()
753 }
754
755 pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
757 self.unit_of_work = Some(config);
758 self
759 }
760
761 pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
763 self.unit_of_work.as_ref()
764 }
765
766 pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
768 self.circuit_breaker.as_ref()
769 }
770
771 pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
772 self.security_policy.as_ref()
773 }
774
775 pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
776 self.security_authenticator.as_ref()
777 }
778
779 pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
781 self.concurrency.as_ref()
782 }
783
784 pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
786 self.concurrency = Some(model);
787 self
788 }
789
790 pub fn route_id(&self) -> &str {
792 &self.route_id
793 }
794
795 pub fn auto_startup(&self) -> bool {
797 self.auto_startup
798 }
799
800 pub fn startup_order(&self) -> i32 {
802 self.startup_order
803 }
804
805 pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
807 self.route_id = id.into();
808 self
809 }
810
811 pub fn with_auto_startup(mut self, auto: bool) -> Self {
813 self.auto_startup = auto;
814 self
815 }
816
817 pub fn with_startup_order(mut self, order: i32) -> Self {
819 self.startup_order = order;
820 self
821 }
822
823 pub fn with_source_hash(mut self, hash: u64) -> Self {
824 self.source_hash = Some(hash);
825 self
826 }
827
828 pub fn source_hash(&self) -> Option<u64> {
829 self.source_hash
830 }
831
832 pub fn to_info(&self) -> RouteDefinitionInfo {
835 RouteDefinitionInfo {
836 route_id: self.route_id.clone(),
837 auto_startup: self.auto_startup,
838 startup_order: self.startup_order,
839 source_hash: self.source_hash,
840 }
841 }
842}
843
844#[derive(Clone)]
850pub struct RouteDefinitionInfo {
851 route_id: String,
852 auto_startup: bool,
853 startup_order: i32,
854 pub(crate) source_hash: Option<u64>,
855}
856
857impl RouteDefinitionInfo {
858 pub fn route_id(&self) -> &str {
860 &self.route_id
861 }
862
863 pub fn auto_startup(&self) -> bool {
865 self.auto_startup
866 }
867
868 pub fn startup_order(&self) -> i32 {
870 self.startup_order
871 }
872
873 pub fn source_hash(&self) -> Option<u64> {
874 self.source_hash
875 }
876}
877
878#[cfg(test)]
879mod tests {
880 use super::*;
881
882 #[test]
889 fn builder_step_span_label_mapping() {
890 use camel_api::declarative::LanguageExpressionDef;
891 use camel_api::splitter::AggregationStrategy;
892 use camel_api::{BoxProcessor, IdentityProcessor, OpaqueProcessor};
893
894 let expr = LanguageExpressionDef {
895 language: "simple".into(),
896 source: "${body}".into(),
897 };
898
899 assert_eq!(
901 BuilderStep::To("direct:tree-sub".into())
902 .span_label()
903 .as_deref(),
904 Some("to:direct")
905 );
906 assert_eq!(
907 BuilderStep::To("http://api.example/x".into())
908 .span_label()
909 .as_deref(),
910 Some("to:http")
911 );
912 assert_eq!(BuilderStep::To("garbage".into()).span_label(), None);
914
915 assert_eq!(
917 BuilderStep::Log {
918 level: camel_processor::LogLevel::Info,
919 message: "m".into(),
920 }
921 .span_label()
922 .as_deref(),
923 Some("log")
924 );
925 assert_eq!(
926 BuilderStep::Split {
927 config: camel_api::splitter::SplitterConfig::new(
928 camel_api::splitter::split_body_lines()
929 ),
930 steps: vec![BuilderStep::Stop],
931 }
932 .span_label()
933 .as_deref(),
934 Some("split")
935 );
936 assert_eq!(
937 BuilderStep::DeclarativeSplit {
938 expression: expr.clone(),
939 aggregation: AggregationStrategy::Original,
940 parallel: false,
941 parallel_limit: None,
942 stop_on_exception: true,
943 steps: vec![BuilderStep::Stop],
944 }
945 .span_label()
946 .as_deref(),
947 Some("split")
948 );
949 assert_eq!(
950 BuilderStep::DeclarativeStreamSplit {
951 stream_config: camel_api::StreamSplitConfig::default(),
952 aggregation: AggregationStrategy::Original,
953 stop_on_exception: true,
954 steps: vec![BuilderStep::Stop],
955 }
956 .span_label()
957 .as_deref(),
958 Some("split")
959 );
960 assert_eq!(BuilderStep::Stop.span_label(), None);
961
962 assert_eq!(
964 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
965 .span_label(),
966 None
967 );
968 }
969
970 #[test]
982 fn builder_step_span_kind_hint_mapping() {
983 use camel_api::splitter::split_body_lines;
984 use camel_api::{Exchange, FilterPredicate, SpanKindHint};
985
986 let kind = |uri: &str| BuilderStep::To(uri.into()).span_kind_hint();
987
988 assert_eq!(kind("kafka:orders"), SpanKindHint::Producer);
990 assert_eq!(kind("jms:q"), SpanKindHint::Producer);
991 assert_eq!(kind("activemq:q"), SpanKindHint::Producer);
992 assert_eq!(kind("artemis:q"), SpanKindHint::Producer);
993 assert_eq!(kind("mqtt:t"), SpanKindHint::Producer);
994 assert_eq!(kind("KAFKA:orders"), SpanKindHint::Producer);
996
997 assert_eq!(kind("http://x"), SpanKindHint::Client);
999 assert_eq!(kind("https://x"), SpanKindHint::Client);
1000 assert_eq!(kind("grpc://x"), SpanKindHint::Client);
1001 assert_eq!(kind("grpcs://x"), SpanKindHint::Client);
1002 assert_eq!(kind("ws://x"), SpanKindHint::Client);
1003 assert_eq!(kind("redis://x"), SpanKindHint::Client);
1004 assert_eq!(kind("opensearch://x"), SpanKindHint::Client);
1005 assert_eq!(kind("sql:db"), SpanKindHint::Client);
1006 assert_eq!(kind("surrealdb://x"), SpanKindHint::Client);
1007 assert_eq!(kind("cxf://x"), SpanKindHint::Client);
1008 assert_eq!(kind("llm://x"), SpanKindHint::Client);
1009 assert_eq!(kind("mcp://x"), SpanKindHint::Client);
1010
1011 assert_eq!(kind("direct:y"), SpanKindHint::Internal);
1013 assert_eq!(kind("timer:z"), SpanKindHint::Internal);
1014 assert_eq!(kind("garbage"), SpanKindHint::Internal);
1016
1017 assert_eq!(
1019 BuilderStep::Log {
1020 level: camel_processor::LogLevel::Info,
1021 message: "m".into(),
1022 }
1023 .span_kind_hint(),
1024 SpanKindHint::Internal
1025 );
1026 assert_eq!(
1027 BuilderStep::Filter {
1028 predicate: FilterPredicate::new(|_: &Exchange| true),
1029 steps: vec![BuilderStep::Stop],
1030 }
1031 .span_kind_hint(),
1032 SpanKindHint::Internal
1033 );
1034 assert_eq!(
1035 BuilderStep::Split {
1036 config: camel_api::splitter::SplitterConfig::new(split_body_lines()),
1037 steps: vec![BuilderStep::Stop],
1038 }
1039 .span_kind_hint(),
1040 SpanKindHint::Internal
1041 );
1042
1043 let enrich = |uri: &str| BuilderStep::Enrich {
1047 uri: uri.into(),
1048 strategy: None,
1049 timeout_ms: None,
1050 };
1051 let poll_enrich = |uri: &str| BuilderStep::PollEnrich {
1052 uri: uri.into(),
1053 strategy: None,
1054 timeout_ms: None,
1055 };
1056 let wire_tap = |uri: &str| BuilderStep::WireTap { uri: uri.into() };
1057
1058 assert_eq!(enrich("http://x").span_kind_hint(), SpanKindHint::Client);
1060 assert_eq!(
1061 poll_enrich("http://x").span_kind_hint(),
1062 SpanKindHint::Client
1063 );
1064 assert_eq!(wire_tap("http://x").span_kind_hint(), SpanKindHint::Client);
1065 assert_eq!(
1067 enrich("kafka:orders").span_kind_hint(),
1068 SpanKindHint::Producer
1069 );
1070 assert_eq!(
1071 poll_enrich("kafka:orders").span_kind_hint(),
1072 SpanKindHint::Producer
1073 );
1074 assert_eq!(
1075 wire_tap("kafka:orders").span_kind_hint(),
1076 SpanKindHint::Producer
1077 );
1078 assert_eq!(enrich("direct:y").span_kind_hint(), SpanKindHint::Internal);
1080 assert_eq!(
1081 poll_enrich("seda:q").span_kind_hint(),
1082 SpanKindHint::Internal
1083 );
1084 assert_eq!(
1085 wire_tap("direct:y").span_kind_hint(),
1086 SpanKindHint::Internal
1087 );
1088 }
1089
1090 #[test]
1095 fn golden_debug_output_all_variants() {
1096 use camel_api::declarative::LanguageExpressionDef;
1097 use camel_api::loop_eip::LoopMode;
1098 use camel_api::recipient_list::RecipientListConfig;
1099 use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
1100 use camel_api::{
1101 BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
1102 FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
1103 Value,
1104 };
1105 use std::sync::Arc;
1106
1107 let expr = LanguageExpressionDef {
1108 language: "simple".into(),
1109 source: "${body}".into(),
1110 };
1111
1112 assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
1115 assert_eq!(
1116 format!(
1117 "{:?}",
1118 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
1119 ),
1120 "Processor(BoxProcessor(...))"
1121 );
1122 assert_eq!(
1123 format!("{:?}", BuilderStep::To("mock:out".into())),
1124 "To(\"mock:out\")"
1125 );
1126
1127 assert_eq!(
1130 format!(
1131 "{:?}",
1132 BuilderStep::Log {
1133 level: camel_processor::LogLevel::Info,
1134 message: "hello".into(),
1135 }
1136 ),
1137 "Log { level: Info, message: \"hello\" }"
1138 );
1139
1140 assert_eq!(
1141 format!(
1142 "{:?}",
1143 BuilderStep::DeclarativeSetHeader {
1144 key: "k".into(),
1145 value: ValueSourceDef::Literal(Value::String("v".into())),
1146 }
1147 ),
1148 "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
1149 );
1150
1151 assert_eq!(
1152 format!(
1153 "{:?}",
1154 BuilderStep::DeclarativeSetHeaderIfAbsent {
1155 key: "k".into(),
1156 value: ValueSourceDef::Literal(Value::String("v".into())),
1157 }
1158 ),
1159 "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
1160 );
1161
1162 assert_eq!(
1163 format!(
1164 "{:?}",
1165 BuilderStep::DeclarativeSetBody {
1166 value: ValueSourceDef::Literal(Value::String("v".into())),
1167 }
1168 ),
1169 "DeclarativeSetBody { value: Literal(String(\"v\")) }"
1170 );
1171
1172 assert_eq!(
1173 format!(
1174 "{:?}",
1175 BuilderStep::DeclarativeSetProperty {
1176 key: "prop".into(),
1177 value_source: ValueSourceDef::Literal(Value::String("v".into())),
1178 }
1179 ),
1180 "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
1181 );
1182
1183 assert_eq!(
1184 format!(
1185 "{:?}",
1186 BuilderStep::DeclarativeScript {
1187 expression: expr.clone(),
1188 }
1189 ),
1190 "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
1191 );
1192
1193 let func_def = FunctionDefinition {
1195 id: FunctionId("test-id".into()),
1196 runtime: "my_runtime".into(),
1197 source: "${body}".into(),
1198 timeout_ms: 5000,
1199 route_id: None,
1200 step_index: None,
1201 };
1202 assert_eq!(
1203 format!(
1204 "{:?}",
1205 BuilderStep::DeclarativeFunction {
1206 definition: func_def,
1207 }
1208 ),
1209 "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
1210 );
1211
1212 assert_eq!(
1213 format!(
1214 "{:?}",
1215 BuilderStep::WireTap {
1216 uri: "mock:tap".into(),
1217 }
1218 ),
1219 "WireTap { uri: \"mock:tap\" }"
1220 );
1221
1222 assert_eq!(
1223 format!(
1224 "{:?}",
1225 BuilderStep::DeclarativeLog {
1226 level: camel_processor::LogLevel::Info,
1227 message: ValueSourceDef::Expression(expr.clone()),
1228 }
1229 ),
1230 "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
1231 );
1232
1233 assert_eq!(
1234 format!(
1235 "{:?}",
1236 BuilderStep::Bean {
1237 name: "myBean".into(),
1238 method: "process".into(),
1239 }
1240 ),
1241 "Bean { name: \"myBean\", method: \"process\" }"
1242 );
1243
1244 assert_eq!(
1245 format!(
1246 "{:?}",
1247 BuilderStep::Script {
1248 language: "js".into(),
1249 script: "body".into(),
1250 }
1251 ),
1252 "Script { language: \"js\", script: \"body\" }"
1253 );
1254
1255 assert_eq!(
1256 format!(
1257 "{:?}",
1258 BuilderStep::Aggregate {
1259 config: camel_api::AggregatorConfig::correlate_by("id")
1260 .complete_when_size(1)
1261 .build()
1262 .unwrap(),
1263 }
1264 ),
1265 "Aggregate { config: AggregatorConfig { header_name: \"id\", completion: Single(Size(1)), correlation: HeaderName(\"id\"), strategy: CollectAll, max_buckets: Some(10000), max_bucket_size: Some(10000), bucket_ttl: Some(300s), force_completion_on_stop: false, discard_on_timeout: false, max_timeout_tasks: 1024 } }"
1266 );
1267
1268 assert_eq!(
1269 format!(
1270 "{:?}",
1271 BuilderStep::DynamicRouter {
1272 config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
1273 "mock:dr".into()
1274 ))),
1275 }
1276 ),
1277 "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
1278 );
1279
1280 assert_eq!(
1281 format!(
1282 "{:?}",
1283 BuilderStep::RoutingSlip {
1284 config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
1285 }
1286 ),
1287 "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
1288 );
1289
1290 assert_eq!(
1291 format!(
1292 "{:?}",
1293 BuilderStep::RecipientList {
1294 config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
1295 }
1296 ),
1297 "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
1298 );
1299
1300 assert_eq!(
1301 format!(
1302 "{:?}",
1303 BuilderStep::Enrich {
1304 uri: "mock:enrich".into(),
1305 strategy: Some("agg".into()),
1306 timeout_ms: Some(1000),
1307 }
1308 ),
1309 "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
1310 );
1311
1312 assert_eq!(
1313 format!(
1314 "{:?}",
1315 BuilderStep::PollEnrich {
1316 uri: "mock:poll".into(),
1317 strategy: None,
1318 timeout_ms: None,
1319 }
1320 ),
1321 "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
1322 );
1323
1324 assert_eq!(
1325 format!(
1326 "{:?}",
1327 BuilderStep::Validate {
1328 predicate: expr.clone(),
1329 }
1330 ),
1331 "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
1332 );
1333
1334 assert_eq!(
1335 format!("{:?}", BuilderStep::Sampling { period: 100 }),
1336 "Sampling { period: 100 }"
1337 );
1338
1339 assert_eq!(
1340 format!(
1341 "{:?}",
1342 BuilderStep::Resequence {
1343 policy_config: Default::default(),
1344 }
1345 ),
1346 "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
1347 );
1348
1349 assert_eq!(
1351 format!(
1352 "{:?}",
1353 BuilderStep::DeclarativeFilter {
1354 predicate: expr.clone(),
1355 steps: vec![BuilderStep::Stop],
1356 }
1357 ),
1358 "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
1359 );
1360
1361 assert_eq!(
1362 format!(
1363 "{:?}",
1364 BuilderStep::DeclarativeSplit {
1365 expression: expr.clone(),
1366 aggregation: AggregationStrategy::Original,
1367 parallel: false,
1368 parallel_limit: Some(2),
1369 stop_on_exception: true,
1370 steps: vec![BuilderStep::Stop],
1371 }
1372 ),
1373 "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
1374 );
1375
1376 assert_eq!(
1377 format!(
1378 "{:?}",
1379 BuilderStep::Split {
1380 config: camel_api::splitter::SplitterConfig::new(
1381 camel_api::splitter::split_body_lines()
1382 ),
1383 steps: vec![BuilderStep::Stop],
1384 }
1385 ),
1386 "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
1387 );
1388
1389 assert_eq!(
1390 format!(
1391 "{:?}",
1392 BuilderStep::Filter {
1393 predicate: FilterPredicate::new(|_: &Exchange| true),
1394 steps: vec![BuilderStep::Stop],
1395 }
1396 ),
1397 "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
1398 );
1399
1400 assert_eq!(
1401 format!(
1402 "{:?}",
1403 BuilderStep::Throttle {
1404 config: camel_api::ThrottlerConfig::new(
1405 10,
1406 std::time::Duration::from_millis(10)
1407 ),
1408 steps: vec![BuilderStep::Stop],
1409 }
1410 ),
1411 "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
1412 );
1413
1414 assert_eq!(
1415 format!(
1416 "{:?}",
1417 BuilderStep::LoadBalance {
1418 config: camel_api::LoadBalancerConfig::round_robin(),
1419 steps: vec![BuilderStep::To("mock:l1".into())],
1420 }
1421 ),
1422 "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
1423 );
1424
1425 assert_eq!(
1426 format!(
1427 "{:?}",
1428 BuilderStep::Delay {
1429 config: camel_api::DelayConfig::new(500),
1430 }
1431 ),
1432 "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
1433 );
1434
1435 assert_eq!(
1439 format!(
1440 "{:?}",
1441 BuilderStep::Choice {
1442 whens: vec![WhenStep {
1443 predicate: FilterPredicate::new(|_: &Exchange| true),
1444 steps: vec![BuilderStep::To("mock:a".into())],
1445 }],
1446 otherwise: None,
1447 }
1448 ),
1449 "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
1450 );
1451
1452 assert_eq!(
1453 format!(
1454 "{:?}",
1455 BuilderStep::DeclarativeChoice {
1456 whens: vec![DeclarativeWhenStep {
1457 predicate: expr.clone(),
1458 steps: vec![BuilderStep::Stop],
1459 }],
1460 otherwise: Some(vec![BuilderStep::Stop]),
1461 }
1462 ),
1463 "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
1464 );
1465
1466 assert_eq!(
1468 format!(
1469 "{:?}",
1470 BuilderStep::Multicast {
1471 steps: vec![BuilderStep::To("direct:a".into())],
1472 config: MulticastConfig::new(),
1473 }
1474 ),
1475 "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
1476 );
1477
1478 assert_eq!(
1480 format!(
1481 "{:?}",
1482 BuilderStep::DeclarativeDynamicRouter {
1483 expression: expr.clone(),
1484 uri_delimiter: ",".into(),
1485 cache_size: 1000,
1486 ignore_invalid_endpoints: false,
1487 max_iterations: 1000,
1488 }
1489 ),
1490 "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
1491 );
1492
1493 assert_eq!(
1494 format!(
1495 "{:?}",
1496 BuilderStep::DeclarativeRoutingSlip {
1497 expression: expr.clone(),
1498 uri_delimiter: ",".into(),
1499 cache_size: 1000,
1500 ignore_invalid_endpoints: false,
1501 }
1502 ),
1503 "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
1504 );
1505
1506 assert_eq!(
1508 format!(
1509 "{:?}",
1510 BuilderStep::DeclarativeRecipientList {
1511 expression: expr.clone(),
1512 delimiter: ",".into(),
1513 parallel: false,
1514 parallel_limit: None,
1515 stop_on_exception: false,
1516 aggregation: "original".into(),
1517 }
1518 ),
1519 "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
1520 );
1521
1522 assert_eq!(
1524 format!(
1525 "{:?}",
1526 BuilderStep::Loop {
1527 config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
1528 steps: vec![],
1529 }
1530 ),
1531 "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
1532 );
1533
1534 assert_eq!(
1535 format!(
1536 "{:?}",
1537 BuilderStep::DeclarativeLoop {
1538 count: Some(5),
1539 while_predicate: None,
1540 steps: vec![],
1541 max_iterations: Some(100),
1542 }
1543 ),
1544 "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
1545 );
1546
1547 assert_eq!(
1549 format!(
1550 "{:?}",
1551 BuilderStep::ClaimCheck {
1552 repository: "myRepo".into(),
1553 operation: "checkout".into(),
1554 key: expr.clone(),
1555 filter: None,
1556 }
1557 ),
1558 "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
1559 );
1560
1561 assert_eq!(
1563 format!(
1564 "{:?}",
1565 BuilderStep::Sort {
1566 expression: expr.clone(),
1567 reverse: false,
1568 }
1569 ),
1570 "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
1571 );
1572
1573 assert_eq!(
1575 format!(
1576 "{:?}",
1577 BuilderStep::IdempotentConsumer {
1578 repository: "myRepo".into(),
1579 expression: expr.clone(),
1580 steps: vec![],
1581 eager: true,
1582 remove_on_failure: false,
1583 }
1584 ),
1585 "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
1586 );
1587
1588 assert_eq!(
1590 format!(
1591 "{:?}",
1592 BuilderStep::DeclarativeDoTry {
1593 try_steps: vec![BuilderStep::Stop],
1594 catch: vec![],
1595 finally: None,
1596 }
1597 ),
1598 "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1599 );
1600
1601 assert_eq!(
1603 format!(
1604 "{:?}",
1605 BuilderStep::DeclarativeStreamSplit {
1606 stream_config: StreamSplitConfig {
1607 format: StreamSplitFormat::Ndjson,
1608 max_record_bytes: 1024 * 1024,
1609 batch_size: 1,
1610 chunk_size: None,
1611 include_origin: true,
1612 },
1613 aggregation: AggregationStrategy::Original,
1614 stop_on_exception: true,
1615 steps: vec![BuilderStep::Stop],
1616 }
1617 ),
1618 "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] }"
1619 );
1620 }
1621
1622 #[test]
1623 fn test_builder_step_multicast_variant() {
1624 use camel_api::MulticastConfig;
1625
1626 let step = BuilderStep::Multicast {
1627 steps: vec![BuilderStep::To("direct:a".into())],
1628 config: MulticastConfig::new(),
1629 };
1630
1631 assert!(matches!(step, BuilderStep::Multicast { .. }));
1632 }
1633
1634 #[test]
1635 fn test_route_definition_defaults() {
1636 let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1637 assert_eq!(def.route_id(), "test-route");
1638 assert!(def.auto_startup());
1639 assert_eq!(def.startup_order(), 1000);
1640 }
1641
1642 #[test]
1643 fn test_route_definition_builders() {
1644 let def = RouteDefinition::new("direct:test", vec![])
1645 .with_route_id("my-route")
1646 .with_auto_startup(false)
1647 .with_startup_order(50);
1648 assert_eq!(def.route_id(), "my-route");
1649 assert!(!def.auto_startup());
1650 assert_eq!(def.startup_order(), 50);
1651 }
1652
1653 #[test]
1654 fn test_route_definition_accessors_cover_core_fields() {
1655 let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1656 .with_route_id("accessor-route");
1657
1658 assert_eq!(def.from_uri(), "direct:in");
1659 assert_eq!(def.steps().len(), 1);
1660 assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1661 }
1662
1663 #[test]
1664 fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1665 use camel_api::circuit_breaker::CircuitBreakerConfig;
1666 use camel_api::error_handler::ErrorHandlerConfig;
1667 use camel_component_api::ConcurrencyModel;
1668
1669 let def = RouteDefinition::new("direct:test", vec![])
1670 .with_route_id("eh-route")
1671 .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1672 .with_circuit_breaker(CircuitBreakerConfig::new())
1673 .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1674
1675 let eh = def
1676 .error_handler_config()
1677 .expect("error handler should be set");
1678 assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1679 assert!(def.circuit_breaker_config().is_some());
1680 assert!(matches!(
1681 def.concurrency_override(),
1682 Some(ConcurrencyModel::Concurrent { max: Some(4) })
1683 ));
1684 }
1685
1686 #[test]
1687 fn test_builder_step_debug_covers_many_variants() {
1688 use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1689 use camel_api::{
1690 BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1691 OpaqueProcessor, RoutingSlipConfig, Value,
1692 };
1693 use std::sync::Arc;
1694
1695 let expr = LanguageExpressionDef {
1696 language: "simple".into(),
1697 source: "${body}".into(),
1698 };
1699
1700 let steps = vec![
1701 BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1702 BuilderStep::To("mock:out".into()),
1703 BuilderStep::Stop,
1704 BuilderStep::Log {
1705 level: camel_processor::LogLevel::Info,
1706 message: "hello".into(),
1707 },
1708 BuilderStep::DeclarativeSetHeader {
1709 key: "k".into(),
1710 value: ValueSourceDef::Literal(Value::String("v".into())),
1711 },
1712 BuilderStep::DeclarativeSetBody {
1713 value: ValueSourceDef::Expression(expr.clone()),
1714 },
1715 BuilderStep::DeclarativeFilter {
1716 predicate: expr.clone(),
1717 steps: vec![BuilderStep::Stop],
1718 },
1719 BuilderStep::DeclarativeChoice {
1720 whens: vec![DeclarativeWhenStep {
1721 predicate: expr.clone(),
1722 steps: vec![BuilderStep::Stop],
1723 }],
1724 otherwise: Some(vec![BuilderStep::Stop]),
1725 },
1726 BuilderStep::DeclarativeScript {
1727 expression: expr.clone(),
1728 },
1729 BuilderStep::DeclarativeSplit {
1730 expression: expr.clone(),
1731 aggregation: AggregationStrategy::Original,
1732 parallel: false,
1733 parallel_limit: Some(2),
1734 stop_on_exception: true,
1735 steps: vec![BuilderStep::Stop],
1736 },
1737 BuilderStep::Split {
1738 config: SplitterConfig::new(split_body_lines()),
1739 steps: vec![BuilderStep::Stop],
1740 },
1741 BuilderStep::Aggregate {
1742 config: camel_api::AggregatorConfig::correlate_by("id")
1743 .complete_when_size(1)
1744 .build()
1745 .unwrap(),
1746 },
1747 BuilderStep::Filter {
1748 predicate: FilterPredicate::new(|_: &Exchange| true),
1749 steps: vec![BuilderStep::Stop],
1750 },
1751 BuilderStep::WireTap {
1752 uri: "mock:tap".into(),
1753 },
1754 BuilderStep::DeclarativeLog {
1755 level: camel_processor::LogLevel::Info,
1756 message: ValueSourceDef::Expression(expr.clone()),
1757 },
1758 BuilderStep::Bean {
1759 name: "bean".into(),
1760 method: "call".into(),
1761 },
1762 BuilderStep::Script {
1763 language: "rhai".into(),
1764 script: "body".into(),
1765 },
1766 BuilderStep::Throttle {
1767 config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1768 steps: vec![BuilderStep::Stop],
1769 },
1770 BuilderStep::LoadBalance {
1771 config: camel_api::LoadBalancerConfig::round_robin(),
1772 steps: vec![BuilderStep::To("mock:l1".into())],
1773 },
1774 BuilderStep::DynamicRouter {
1775 config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1776 },
1777 BuilderStep::RoutingSlip {
1778 config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1779 },
1780 ];
1781
1782 for step in steps {
1783 let dbg = format!("{step:?}");
1784 assert!(!dbg.is_empty());
1785 }
1786 }
1787
1788 #[test]
1789 fn test_route_definition_to_info_preserves_metadata() {
1790 let info = RouteDefinition::new("direct:test", vec![])
1791 .with_route_id("meta-route")
1792 .with_auto_startup(false)
1793 .with_startup_order(7)
1794 .to_info();
1795
1796 assert_eq!(info.route_id(), "meta-route");
1797 assert!(!info.auto_startup());
1798 assert_eq!(info.startup_order(), 7);
1799 }
1800
1801 #[test]
1802 fn test_choice_builder_step_debug() {
1803 use camel_api::FilterPredicate;
1804
1805 fn always_true(_: &camel_api::Exchange) -> bool {
1806 true
1807 }
1808
1809 let step = BuilderStep::Choice {
1810 whens: vec![WhenStep {
1811 predicate: FilterPredicate::new(always_true),
1812 steps: vec![BuilderStep::To("mock:a".into())],
1813 }],
1814 otherwise: None,
1815 };
1816 let debug = format!("{step:?}");
1817 assert!(debug.contains("Choice"));
1818 }
1819
1820 #[test]
1821 fn test_route_definition_unit_of_work() {
1822 use camel_api::UnitOfWorkConfig;
1823 let config = UnitOfWorkConfig {
1824 on_complete: Some("log:complete".into()),
1825 on_failure: Some("log:failed".into()),
1826 };
1827 let def = RouteDefinition::new("direct:test", vec![])
1828 .with_route_id("uow-test")
1829 .with_unit_of_work(config.clone());
1830 assert_eq!(
1831 def.unit_of_work_config().unwrap().on_complete.as_deref(),
1832 Some("log:complete")
1833 );
1834 assert_eq!(
1835 def.unit_of_work_config().unwrap().on_failure.as_deref(),
1836 Some("log:failed")
1837 );
1838
1839 let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1840 assert!(def_no_uow.unit_of_work_config().is_none());
1841 }
1842
1843 #[test]
1844 fn test_route_definition_security_policy_accessor() {
1845 use async_trait::async_trait;
1846 use camel_api::CamelError;
1847 use camel_api::Exchange;
1848 use camel_api::security_policy::{
1849 AuthContext, AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1850 };
1851
1852 struct StubPolicy;
1853 #[async_trait]
1854 impl SecurityPolicy for StubPolicy {
1855 async fn evaluate(
1856 &self,
1857 _exchange: &mut Exchange,
1858 _auth: &AuthContext<'_>,
1859 ) -> Result<AuthorizationDecision, CamelError> {
1860 Ok(AuthorizationDecision::Granted {
1861 principal: Principal {
1862 subject: "test".into(),
1863 issuer: "test".into(),
1864 audience: vec![],
1865 scopes: vec![],
1866 roles: vec![],
1867 claims: serde_json::Value::Null,
1868 },
1869 })
1870 }
1871 }
1872
1873 let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1874 assert!(def_no_sp.security_policy_config().is_none());
1875
1876 let def = RouteDefinition::new("direct:test", vec![])
1877 .with_route_id("sp-test")
1878 .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1879 assert!(def.security_policy_config().is_some());
1880 }
1881
1882 #[test]
1883 fn test_route_definition_security_authenticator_accessor() {
1884 use camel_api::security_policy::Principal;
1885
1886 struct TestAuth;
1887 #[async_trait::async_trait]
1888 impl TokenAuthenticator for TestAuth {
1889 async fn authenticate_bearer(
1890 &self,
1891 _token: &str,
1892 ) -> Result<Principal, camel_api::CamelError> {
1893 Ok(Principal {
1894 subject: "test".into(),
1895 issuer: "test".into(),
1896 audience: vec![],
1897 scopes: vec![],
1898 roles: vec![],
1899 claims: serde_json::Value::Null,
1900 })
1901 }
1902 }
1903
1904 let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1905 assert!(def_no_auth.security_authenticator().is_none());
1906
1907 let auth = Arc::new(TestAuth);
1908 let def = RouteDefinition::new("direct:test".to_string(), vec![])
1909 .with_security_authenticator(auth);
1910 assert!(def.security_authenticator().is_some());
1911 }
1912
1913 #[test]
1914 fn test_map_steps_swaps_steps_and_preserves_other_fields() {
1915 let original = RouteDefinition::new(
1916 "direct:test".to_string(),
1917 vec![BuilderStep::To("mock:a".into()), BuilderStep::Stop],
1918 )
1919 .with_route_id("my-route");
1920
1921 let mapped = original.map_steps(|steps| {
1922 let mut out = Vec::with_capacity(steps.len() + 1);
1923 out.push(BuilderStep::To("mock:prefix".into()));
1924 out.extend(steps);
1925 out
1926 });
1927
1928 assert_eq!(mapped.steps().len(), 3);
1930 assert!(matches!(mapped.steps()[0], BuilderStep::To(ref s) if s == "mock:prefix"));
1931 assert!(matches!(mapped.steps()[1], BuilderStep::To(ref s) if s == "mock:a"));
1932 assert_eq!(mapped.route_id(), "my-route");
1934 }
1935
1936 #[test]
1937 fn circuit_breaker_fallback_accessor_returns_steps() {
1938 let def = RouteDefinition::new("direct:start", vec![])
1939 .with_circuit_breaker_fallback(vec![BuilderStep::To("mock:out".into())]);
1940 assert_eq!(def.circuit_breaker_fallback().len(), 1);
1941 }
1942}