camel_core/lifecycle/application/
route_definition.rs

1// lifecycle/application/route_definition.rs
2// Route definition and builder-step types. Route (compiled artifact) lives in adapters.
3
4use 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::{AggregatorConfig, BoxProcessor, FilterPredicate, MulticastConfig, SplitterConfig};
12use camel_auth::TokenAuthenticator;
13use camel_component_api::ConcurrencyModel;
14
15/// An unresolved when-clause: predicate + nested steps for the sub-pipeline.
16pub struct WhenStep {
17    pub predicate: FilterPredicate,
18    pub steps: Vec<BuilderStep>,
19}
20
21pub use camel_api::declarative::{LanguageExpressionDef, ValueSourceDef};
22
23/// Declarative `when` clause resolved later by the runtime.
24#[derive(Debug)]
25pub struct DeclarativeWhenStep {
26    pub predicate: LanguageExpressionDef,
27    pub steps: Vec<BuilderStep>,
28}
29
30/// A step in an unresolved route definition.
31pub enum BuilderStep {
32    /// A pre-built Tower processor service.
33    Processor(BoxProcessor),
34    /// A destination URI — resolved at start time by CamelContext.
35    To(String),
36    /// A stop step that halts processing immediately.
37    Stop,
38    /// A static log step.
39    Log {
40        level: camel_processor::LogLevel,
41        message: String,
42    },
43    /// Declarative set_header (literal or language-based value), resolved at route-add time.
44    DeclarativeSetHeader {
45        key: String,
46        value: ValueSourceDef,
47    },
48    DeclarativeSetProperty {
49        key: String,
50        value_source: ValueSourceDef,
51    },
52    /// Declarative set_body (literal or language-based value), resolved at route-add time.
53    DeclarativeSetBody {
54        value: ValueSourceDef,
55    },
56    /// Declarative filter using a language predicate, resolved at route-add time.
57    DeclarativeFilter {
58        predicate: LanguageExpressionDef,
59        steps: Vec<BuilderStep>,
60    },
61    /// Declarative choice/when/otherwise using language predicates, resolved at route-add time.
62    DeclarativeChoice {
63        whens: Vec<DeclarativeWhenStep>,
64        otherwise: Option<Vec<BuilderStep>>,
65    },
66    /// Declarative script step evaluated by language and written to body.
67    DeclarativeScript {
68        expression: LanguageExpressionDef,
69    },
70    DeclarativeFunction {
71        definition: camel_api::FunctionDefinition,
72    },
73    /// Declarative split using a language expression, resolved at route-add time.
74    DeclarativeSplit {
75        expression: LanguageExpressionDef,
76        aggregation: camel_api::splitter::AggregationStrategy,
77        parallel: bool,
78        parallel_limit: Option<usize>,
79        stop_on_exception: bool,
80        steps: Vec<BuilderStep>,
81    },
82    DeclarativeDynamicRouter {
83        expression: LanguageExpressionDef,
84        uri_delimiter: String,
85        cache_size: i32,
86        ignore_invalid_endpoints: bool,
87        max_iterations: usize,
88    },
89    DeclarativeRoutingSlip {
90        expression: LanguageExpressionDef,
91        uri_delimiter: String,
92        cache_size: i32,
93        ignore_invalid_endpoints: bool,
94    },
95    /// A Splitter sub-pipeline: config + nested steps to execute per fragment.
96    Split {
97        config: SplitterConfig,
98        steps: Vec<BuilderStep>,
99    },
100    /// An Aggregator step: collects exchanges by correlation key, emits when complete.
101    Aggregate {
102        config: AggregatorConfig,
103    },
104    /// A Filter sub-pipeline: predicate + nested steps executed only when predicate is true.
105    Filter {
106        predicate: FilterPredicate,
107        steps: Vec<BuilderStep>,
108    },
109    /// A Choice step: evaluates when-clauses in order, routes to the first match.
110    /// If no when matches, the optional otherwise branch is used.
111    Choice {
112        whens: Vec<WhenStep>,
113        otherwise: Option<Vec<BuilderStep>>,
114    },
115    /// A WireTap step: sends a clone of the exchange to a tap endpoint (fire-and-forget).
116    WireTap {
117        uri: String,
118    },
119    /// A Multicast step: sends the same exchange to multiple destinations.
120    Multicast {
121        steps: Vec<BuilderStep>,
122        config: MulticastConfig,
123    },
124    /// Declarative log step with a language-evaluated message, resolved at route-add time.
125    DeclarativeLog {
126        level: camel_processor::LogLevel,
127        message: ValueSourceDef,
128    },
129    /// Bean invocation step — resolved at route-add time.
130    Bean {
131        name: String,
132        method: String,
133    },
134    /// Script step: executes a script that can mutate the exchange.
135    /// The script has access to `headers`, `properties`, and `body`.
136    Script {
137        language: String,
138        script: String,
139    },
140    /// Throttle step: rate limiting with configurable behavior when limit exceeded.
141    Throttle {
142        config: camel_api::ThrottlerConfig,
143        steps: Vec<BuilderStep>,
144    },
145    /// LoadBalance step: distributes exchanges across multiple endpoints using a strategy.
146    LoadBalance {
147        config: camel_api::LoadBalancerConfig,
148        steps: Vec<BuilderStep>,
149    },
150    /// DynamicRouter step: routes exchanges dynamically based on expression evaluation.
151    DynamicRouter {
152        config: camel_api::DynamicRouterConfig,
153    },
154    RoutingSlip {
155        config: camel_api::RoutingSlipConfig,
156    },
157    RecipientList {
158        config: camel_api::recipient_list::RecipientListConfig,
159    },
160    DeclarativeRecipientList {
161        expression: LanguageExpressionDef,
162        delimiter: String,
163        parallel: bool,
164        parallel_limit: Option<usize>,
165        stop_on_exception: bool,
166        aggregation: String,
167    },
168    Delay {
169        config: camel_api::DelayConfig,
170    },
171    /// Runtime loop with closure-based predicate (programmatic DSL).
172    Loop {
173        config: LoopConfig,
174        steps: Vec<BuilderStep>,
175    },
176    /// Declarative loop with optional language-based while predicate (YAML DSL).
177    DeclarativeLoop {
178        count: Option<usize>,
179        while_predicate: Option<LanguageExpressionDef>,
180        steps: Vec<BuilderStep>,
181    },
182}
183
184impl std::fmt::Debug for BuilderStep {
185    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        match self {
187            BuilderStep::Processor(_) => write!(f, "BuilderStep::Processor(...)"),
188            BuilderStep::To(uri) => write!(f, "BuilderStep::To({uri:?})"),
189            BuilderStep::Stop => write!(f, "BuilderStep::Stop"),
190            BuilderStep::Log { level, message } => write!(
191                f,
192                "BuilderStep::Log {{ level: {level:?}, message: {message:?} }}"
193            ),
194            BuilderStep::DeclarativeSetHeader { key, .. } => {
195                write!(
196                    f,
197                    "BuilderStep::DeclarativeSetHeader {{ key: {key:?}, .. }}"
198                )
199            }
200            BuilderStep::DeclarativeSetBody { .. } => {
201                write!(f, "BuilderStep::DeclarativeSetBody {{ .. }}")
202            }
203            BuilderStep::DeclarativeSetProperty { key, .. } => {
204                write!(
205                    f,
206                    "BuilderStep::DeclarativeSetProperty {{ key: {key:?}, .. }}"
207                )
208            }
209            BuilderStep::DeclarativeFilter { steps, .. } => {
210                write!(
211                    f,
212                    "BuilderStep::DeclarativeFilter {{ steps: {steps:?}, .. }}"
213                )
214            }
215            BuilderStep::DeclarativeChoice { whens, otherwise } => {
216                write!(
217                    f,
218                    "BuilderStep::DeclarativeChoice {{ whens: {} clause(s), otherwise: {} }}",
219                    whens.len(),
220                    if otherwise.is_some() { "Some" } else { "None" }
221                )
222            }
223            BuilderStep::DeclarativeScript { expression } => write!(
224                f,
225                "BuilderStep::DeclarativeScript {{ language: {:?}, .. }}",
226                expression.language
227            ),
228            BuilderStep::DeclarativeFunction { definition } => write!(
229                f,
230                "BuilderStep::DeclarativeFunction {{ id: {:?}, runtime: {:?}, .. }}",
231                definition.id, definition.runtime
232            ),
233            BuilderStep::DeclarativeSplit { steps, .. } => {
234                write!(
235                    f,
236                    "BuilderStep::DeclarativeSplit {{ steps: {steps:?}, .. }}"
237                )
238            }
239            BuilderStep::DeclarativeDynamicRouter { expression, .. } => write!(
240                f,
241                "BuilderStep::DeclarativeDynamicRouter {{ language: {:?}, .. }}",
242                expression.language
243            ),
244            BuilderStep::DeclarativeRoutingSlip { expression, .. } => write!(
245                f,
246                "BuilderStep::DeclarativeRoutingSlip {{ language: {:?}, .. }}",
247                expression.language
248            ),
249            BuilderStep::Split { steps, .. } => {
250                write!(f, "BuilderStep::Split {{ steps: {steps:?}, .. }}")
251            }
252            BuilderStep::Aggregate { .. } => write!(f, "BuilderStep::Aggregate {{ .. }}"),
253            BuilderStep::Filter { steps, .. } => {
254                write!(f, "BuilderStep::Filter {{ steps: {steps:?}, .. }}")
255            }
256            BuilderStep::Choice { whens, otherwise } => {
257                write!(
258                    f,
259                    "BuilderStep::Choice {{ whens: {} clause(s), otherwise: {} }}",
260                    whens.len(),
261                    if otherwise.is_some() { "Some" } else { "None" }
262                )
263            }
264            BuilderStep::WireTap { uri } => write!(f, "BuilderStep::WireTap {{ uri: {uri:?} }}"),
265            BuilderStep::Multicast { steps, .. } => {
266                write!(f, "BuilderStep::Multicast {{ steps: {steps:?}, .. }}")
267            }
268            BuilderStep::DeclarativeLog { level, .. } => {
269                write!(f, "BuilderStep::DeclarativeLog {{ level: {level:?}, .. }}")
270            }
271            BuilderStep::Bean { name, method } => {
272                write!(
273                    f,
274                    "BuilderStep::Bean {{ name: {name:?}, method: {method:?} }}"
275                )
276            }
277            BuilderStep::Script { language, .. } => {
278                write!(f, "BuilderStep::Script {{ language: {language:?}, .. }}")
279            }
280            BuilderStep::Throttle { steps, .. } => {
281                write!(f, "BuilderStep::Throttle {{ steps: {steps:?}, .. }}")
282            }
283            BuilderStep::LoadBalance { steps, .. } => {
284                write!(f, "BuilderStep::LoadBalance {{ steps: {steps:?}, .. }}")
285            }
286            BuilderStep::DynamicRouter { .. } => {
287                write!(f, "BuilderStep::DynamicRouter {{ .. }}")
288            }
289            BuilderStep::RoutingSlip { .. } => {
290                write!(f, "BuilderStep::RoutingSlip {{ .. }}")
291            }
292            BuilderStep::RecipientList { .. } => {
293                write!(f, "BuilderStep::RecipientList {{ .. }}")
294            }
295            BuilderStep::DeclarativeRecipientList {
296                expression,
297                aggregation,
298                ..
299            } => write!(
300                f,
301                "BuilderStep::DeclarativeRecipientList {{ language: {:?}, aggregation: {:?}, .. }}",
302                expression.language, aggregation
303            ),
304            BuilderStep::Delay { config } => {
305                write!(f, "BuilderStep::Delay {{ config: {:?} }}", config)
306            }
307            BuilderStep::Loop { config, steps } => {
308                write!(
309                    f,
310                    "BuilderStep::Loop {{ config: {:?}, steps: {} }}",
311                    config.mode_name(),
312                    steps.len()
313                )
314            }
315            BuilderStep::DeclarativeLoop {
316                count,
317                while_predicate,
318                steps,
319            } => {
320                write!(
321                    f,
322                    "BuilderStep::DeclarativeLoop {{ count: {:?}, while: {}, steps: {} }}",
323                    count,
324                    while_predicate.is_some(),
325                    steps.len()
326                )
327            }
328        }
329    }
330}
331
332/// An unresolved route definition. "to" URIs have not been resolved to producers yet.
333pub struct RouteDefinition {
334    pub(crate) from_uri: String,
335    pub(crate) steps: Vec<BuilderStep>,
336    /// Optional per-route error handler config. Takes precedence over the global one.
337    pub(crate) error_handler: Option<ErrorHandlerConfig>,
338    /// Optional circuit breaker config. Applied between error handler and step pipeline.
339    pub(crate) circuit_breaker: Option<CircuitBreakerConfig>,
340    pub(crate) security_policy: Option<SecurityPolicyConfig>,
341    /// Optional token authenticator for validating JWT/OAuth tokens.
342    pub(crate) security_authenticator: Option<Arc<dyn TokenAuthenticator>>,
343    /// Optional Unit of Work config for in-flight tracking and completion hooks.
344    pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
345    /// User override for the consumer's concurrency model. `None` means
346    /// "use whatever the consumer declares".
347    pub(crate) concurrency: Option<ConcurrencyModel>,
348    /// Unique identifier for this route. Required.
349    pub(crate) route_id: String,
350    /// Whether this route should start automatically when the context starts.
351    pub(crate) auto_startup: bool,
352    /// Order in which routes are started. Lower values start first.
353    pub(crate) startup_order: i32,
354    pub(crate) source_hash: Option<u64>,
355}
356
357impl RouteDefinition {
358    /// Create a new route definition with the required route ID.
359    pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
360        Self {
361            from_uri: from_uri.into(),
362            steps,
363            error_handler: None,
364            circuit_breaker: None,
365            security_policy: None,
366            security_authenticator: None,
367            unit_of_work: None,
368            concurrency: None,
369            route_id: String::new(), // Will be set by with_route_id()
370            auto_startup: true,
371            startup_order: 1000,
372            source_hash: None,
373        }
374    }
375
376    /// The source endpoint URI.
377    pub fn from_uri(&self) -> &str {
378        &self.from_uri
379    }
380
381    /// The steps in this route definition.
382    pub fn steps(&self) -> &[BuilderStep] {
383        &self.steps
384    }
385
386    /// Set a per-route error handler, overriding the global one.
387    pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
388        self.error_handler = Some(config);
389        self
390    }
391
392    /// Get the route-level error handler config, if set.
393    pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
394        self.error_handler.as_ref()
395    }
396
397    /// Set a circuit breaker for this route.
398    pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
399        self.circuit_breaker = Some(config);
400        self
401    }
402
403    /// Set a security policy for this route.
404    pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
405        self.security_policy = Some(config);
406        self
407    }
408
409    /// Set a token authenticator for this route.
410    pub fn with_security_authenticator(
411        mut self,
412        authenticator: Arc<dyn TokenAuthenticator>,
413    ) -> Self {
414        self.security_authenticator = Some(authenticator);
415        self
416    }
417
418    /// Set a unit of work config for this route.
419    pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
420        self.unit_of_work = Some(config);
421        self
422    }
423
424    /// Get the unit of work config, if set.
425    pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
426        self.unit_of_work.as_ref()
427    }
428
429    /// Get the circuit breaker config, if set.
430    pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
431        self.circuit_breaker.as_ref()
432    }
433
434    pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
435        self.security_policy.as_ref()
436    }
437
438    pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
439        self.security_authenticator.as_ref()
440    }
441
442    /// User-specified concurrency override, if any.
443    pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
444        self.concurrency.as_ref()
445    }
446
447    /// Override the consumer's concurrency model for this route.
448    pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
449        self.concurrency = Some(model);
450        self
451    }
452
453    /// Get the route ID.
454    pub fn route_id(&self) -> &str {
455        &self.route_id
456    }
457
458    /// Whether this route should start automatically when the context starts.
459    pub fn auto_startup(&self) -> bool {
460        self.auto_startup
461    }
462
463    /// Order in which routes are started. Lower values start first.
464    pub fn startup_order(&self) -> i32 {
465        self.startup_order
466    }
467
468    /// Set a unique identifier for this route.
469    pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
470        self.route_id = id.into();
471        self
472    }
473
474    /// Set whether this route should start automatically.
475    pub fn with_auto_startup(mut self, auto: bool) -> Self {
476        self.auto_startup = auto;
477        self
478    }
479
480    /// Set the startup order. Lower values start first.
481    pub fn with_startup_order(mut self, order: i32) -> Self {
482        self.startup_order = order;
483        self
484    }
485
486    pub fn with_source_hash(mut self, hash: u64) -> Self {
487        self.source_hash = Some(hash);
488        self
489    }
490
491    pub fn source_hash(&self) -> Option<u64> {
492        self.source_hash
493    }
494
495    /// Extract the metadata fields needed for introspection.
496    /// This is used by RouteController to store route info without the non-Sync steps.
497    pub fn to_info(&self) -> RouteDefinitionInfo {
498        RouteDefinitionInfo {
499            route_id: self.route_id.clone(),
500            auto_startup: self.auto_startup,
501            startup_order: self.startup_order,
502            source_hash: self.source_hash,
503        }
504    }
505}
506
507/// Minimal route definition metadata for introspection.
508///
509/// This struct contains only the metadata fields from [`RouteDefinition`]
510/// that are needed for route lifecycle management, without the `steps` field
511/// (which contains non-Sync types and cannot be stored in a Sync struct).
512#[derive(Clone)]
513pub struct RouteDefinitionInfo {
514    route_id: String,
515    auto_startup: bool,
516    startup_order: i32,
517    pub(crate) source_hash: Option<u64>,
518}
519
520impl RouteDefinitionInfo {
521    /// Get the route ID.
522    pub fn route_id(&self) -> &str {
523        &self.route_id
524    }
525
526    /// Whether this route should start automatically when the context starts.
527    pub fn auto_startup(&self) -> bool {
528        self.auto_startup
529    }
530
531    /// Order in which routes are started. Lower values start first.
532    pub fn startup_order(&self) -> i32 {
533        self.startup_order
534    }
535
536    pub fn source_hash(&self) -> Option<u64> {
537        self.source_hash
538    }
539}
540
541#[cfg(test)]
542mod tests {
543    use super::*;
544
545    #[test]
546    fn test_builder_step_multicast_variant() {
547        use camel_api::MulticastConfig;
548
549        let step = BuilderStep::Multicast {
550            steps: vec![BuilderStep::To("direct:a".into())],
551            config: MulticastConfig::new(),
552        };
553
554        assert!(matches!(step, BuilderStep::Multicast { .. }));
555    }
556
557    #[test]
558    fn test_route_definition_defaults() {
559        let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
560        assert_eq!(def.route_id(), "test-route");
561        assert!(def.auto_startup());
562        assert_eq!(def.startup_order(), 1000);
563    }
564
565    #[test]
566    fn test_route_definition_builders() {
567        let def = RouteDefinition::new("direct:test", vec![])
568            .with_route_id("my-route")
569            .with_auto_startup(false)
570            .with_startup_order(50);
571        assert_eq!(def.route_id(), "my-route");
572        assert!(!def.auto_startup());
573        assert_eq!(def.startup_order(), 50);
574    }
575
576    #[test]
577    fn test_route_definition_accessors_cover_core_fields() {
578        let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
579            .with_route_id("accessor-route");
580
581        assert_eq!(def.from_uri(), "direct:in");
582        assert_eq!(def.steps().len(), 1);
583        assert!(matches!(def.steps()[0], BuilderStep::To(_)));
584    }
585
586    #[test]
587    fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
588        use camel_api::circuit_breaker::CircuitBreakerConfig;
589        use camel_api::error_handler::ErrorHandlerConfig;
590        use camel_component_api::ConcurrencyModel;
591
592        let def = RouteDefinition::new("direct:test", vec![])
593            .with_route_id("eh-route")
594            .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
595            .with_circuit_breaker(CircuitBreakerConfig::new())
596            .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
597
598        let eh = def
599            .error_handler_config()
600            .expect("error handler should be set");
601        assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
602        assert!(def.circuit_breaker_config().is_some());
603        assert!(matches!(
604            def.concurrency_override(),
605            Some(ConcurrencyModel::Concurrent { max: Some(4) })
606        ));
607    }
608
609    #[test]
610    fn test_builder_step_debug_covers_many_variants() {
611        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
612        use camel_api::{
613            DynamicRouterConfig, Exchange, IdentityProcessor, RoutingSlipConfig, Value,
614        };
615        use std::sync::Arc;
616
617        let expr = LanguageExpressionDef {
618            language: "simple".into(),
619            source: "${body}".into(),
620        };
621
622        let steps = vec![
623            BuilderStep::Processor(BoxProcessor::new(IdentityProcessor)),
624            BuilderStep::To("mock:out".into()),
625            BuilderStep::Stop,
626            BuilderStep::Log {
627                level: camel_processor::LogLevel::Info,
628                message: "hello".into(),
629            },
630            BuilderStep::DeclarativeSetHeader {
631                key: "k".into(),
632                value: ValueSourceDef::Literal(Value::String("v".into())),
633            },
634            BuilderStep::DeclarativeSetBody {
635                value: ValueSourceDef::Expression(expr.clone()),
636            },
637            BuilderStep::DeclarativeFilter {
638                predicate: expr.clone(),
639                steps: vec![BuilderStep::Stop],
640            },
641            BuilderStep::DeclarativeChoice {
642                whens: vec![DeclarativeWhenStep {
643                    predicate: expr.clone(),
644                    steps: vec![BuilderStep::Stop],
645                }],
646                otherwise: Some(vec![BuilderStep::Stop]),
647            },
648            BuilderStep::DeclarativeScript {
649                expression: expr.clone(),
650            },
651            BuilderStep::DeclarativeSplit {
652                expression: expr.clone(),
653                aggregation: AggregationStrategy::Original,
654                parallel: false,
655                parallel_limit: Some(2),
656                stop_on_exception: true,
657                steps: vec![BuilderStep::Stop],
658            },
659            BuilderStep::Split {
660                config: SplitterConfig::new(split_body_lines()),
661                steps: vec![BuilderStep::Stop],
662            },
663            BuilderStep::Aggregate {
664                config: camel_api::AggregatorConfig::correlate_by("id")
665                    .complete_when_size(1)
666                    .build()
667                    .unwrap(),
668            },
669            BuilderStep::Filter {
670                predicate: Arc::new(|_: &Exchange| true),
671                steps: vec![BuilderStep::Stop],
672            },
673            BuilderStep::WireTap {
674                uri: "mock:tap".into(),
675            },
676            BuilderStep::DeclarativeLog {
677                level: camel_processor::LogLevel::Info,
678                message: ValueSourceDef::Expression(expr.clone()),
679            },
680            BuilderStep::Bean {
681                name: "bean".into(),
682                method: "call".into(),
683            },
684            BuilderStep::Script {
685                language: "rhai".into(),
686                script: "body".into(),
687            },
688            BuilderStep::Throttle {
689                config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
690                steps: vec![BuilderStep::Stop],
691            },
692            BuilderStep::LoadBalance {
693                config: camel_api::LoadBalancerConfig::round_robin(),
694                steps: vec![BuilderStep::To("mock:l1".into())],
695            },
696            BuilderStep::DynamicRouter {
697                config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
698            },
699            BuilderStep::RoutingSlip {
700                config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
701            },
702        ];
703
704        for step in steps {
705            let dbg = format!("{step:?}");
706            assert!(!dbg.is_empty());
707        }
708    }
709
710    #[test]
711    fn test_route_definition_to_info_preserves_metadata() {
712        let info = RouteDefinition::new("direct:test", vec![])
713            .with_route_id("meta-route")
714            .with_auto_startup(false)
715            .with_startup_order(7)
716            .to_info();
717
718        assert_eq!(info.route_id(), "meta-route");
719        assert!(!info.auto_startup());
720        assert_eq!(info.startup_order(), 7);
721    }
722
723    #[test]
724    fn test_choice_builder_step_debug() {
725        use camel_api::{Exchange, FilterPredicate};
726        use std::sync::Arc;
727
728        fn always_true(_: &Exchange) -> bool {
729            true
730        }
731
732        let step = BuilderStep::Choice {
733            whens: vec![WhenStep {
734                predicate: Arc::new(always_true) as FilterPredicate,
735                steps: vec![BuilderStep::To("mock:a".into())],
736            }],
737            otherwise: None,
738        };
739        let debug = format!("{step:?}");
740        assert!(debug.contains("Choice"));
741    }
742
743    #[test]
744    fn test_route_definition_unit_of_work() {
745        use camel_api::UnitOfWorkConfig;
746        let config = UnitOfWorkConfig {
747            on_complete: Some("log:complete".into()),
748            on_failure: Some("log:failed".into()),
749        };
750        let def = RouteDefinition::new("direct:test", vec![])
751            .with_route_id("uow-test")
752            .with_unit_of_work(config.clone());
753        assert_eq!(
754            def.unit_of_work_config().unwrap().on_complete.as_deref(),
755            Some("log:complete")
756        );
757        assert_eq!(
758            def.unit_of_work_config().unwrap().on_failure.as_deref(),
759            Some("log:failed")
760        );
761
762        let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
763        assert!(def_no_uow.unit_of_work_config().is_none());
764    }
765
766    #[test]
767    fn test_route_definition_security_policy_accessor() {
768        use async_trait::async_trait;
769        use camel_api::CamelError;
770        use camel_api::Exchange;
771        use camel_api::security_policy::{
772            AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
773        };
774
775        struct StubPolicy;
776        #[async_trait]
777        impl SecurityPolicy for StubPolicy {
778            async fn evaluate(
779                &self,
780                _exchange: &mut Exchange,
781            ) -> Result<AuthorizationDecision, CamelError> {
782                Ok(AuthorizationDecision::Granted {
783                    principal: Principal {
784                        subject: "test".into(),
785                        issuer: "test".into(),
786                        audience: vec![],
787                        scopes: vec![],
788                        roles: vec![],
789                        claims: serde_json::Value::Null,
790                    },
791                })
792            }
793        }
794
795        let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
796        assert!(def_no_sp.security_policy_config().is_none());
797
798        let def = RouteDefinition::new("direct:test", vec![])
799            .with_route_id("sp-test")
800            .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
801        assert!(def.security_policy_config().is_some());
802    }
803
804    #[test]
805    fn test_route_definition_security_authenticator_accessor() {
806        use camel_api::security_policy::Principal;
807
808        struct TestAuth;
809        #[async_trait::async_trait]
810        impl TokenAuthenticator for TestAuth {
811            async fn authenticate_bearer(
812                &self,
813                _token: &str,
814            ) -> Result<Principal, camel_api::CamelError> {
815                Ok(Principal {
816                    subject: "test".into(),
817                    issuer: "test".into(),
818                    audience: vec![],
819                    scopes: vec![],
820                    roles: vec![],
821                    claims: serde_json::Value::Null,
822                })
823            }
824        }
825
826        let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
827        assert!(def_no_auth.security_authenticator().is_none());
828
829        let auth = Arc::new(TestAuth);
830        let def = RouteDefinition::new("direct:test".to_string(), vec![])
831            .with_security_authenticator(auth);
832        assert!(def.security_authenticator().is_some());
833    }
834}
camel_core/lifecycle/application/route_definition.rs

camel_core/lifecycle/application/
route_definition.rs
