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

camel_core/lifecycle/application/
route_definition.rs
