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, FilterPredicate, MulticastConfig, OpaqueProcessor, 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
24impl std::fmt::Debug for WhenStep {
25    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        f.debug_struct("WhenStep")
27            .field("predicate", &self.predicate)
28            .field("steps", &self.steps)
29            .finish()
30    }
31}
32
33pub use camel_api::declarative::{LanguageExpressionDef, ValueSourceDef};
34
35/// Declarative `when` clause resolved later by the runtime.
36#[derive(Debug)]
37pub struct DeclarativeWhenStep {
38    pub predicate: LanguageExpressionDef,
39    pub steps: Vec<BuilderStep>,
40}
41
42/// Builder struct for a single `doCatch` clause in the declarative pipeline.
43#[derive(Debug)]
44pub struct DoTryCatchClauseBuilder {
45    pub exception: Option<Vec<String>>,
46    pub when: Option<LanguageExpressionDef>,
47    pub on_when: Option<LanguageExpressionDef>,
48    pub disposition: camel_api::error_handler::ExceptionDisposition,
49    pub steps: Vec<BuilderStep>,
50}
51
52/// Builder struct for the `doFinally` block in the declarative pipeline.
53#[derive(Debug)]
54pub struct DoTryFinallyBuilder {
55    pub on_when: Option<LanguageExpressionDef>,
56    pub steps: Vec<BuilderStep>,
57}
58
59/// A step in an unresolved route definition.
60#[derive(Debug)]
61pub enum BuilderStep {
62    /// A pre-built Tower processor service.
63    Processor(OpaqueProcessor),
64    /// A destination URI — resolved at start time by CamelContext.
65    To(String),
66    /// A stop step that halts processing immediately.
67    Stop,
68    /// A static log step.
69    Log {
70        level: camel_processor::LogLevel,
71        message: String,
72    },
73    /// Declarative set_header (literal or language-based value), resolved at route-add time.
74    DeclarativeSetHeader {
75        key: String,
76        value: ValueSourceDef,
77    },
78    /// Declarative set_header_if_absent (internal-only), resolved at route-add time.
79    DeclarativeSetHeaderIfAbsent {
80        key: String,
81        value: ValueSourceDef,
82    },
83    DeclarativeSetProperty {
84        key: String,
85        value_source: ValueSourceDef,
86    },
87    /// Declarative set_body (literal or language-based value), resolved at route-add time.
88    DeclarativeSetBody {
89        value: ValueSourceDef,
90    },
91    /// Declarative filter using a language predicate, resolved at route-add time.
92    DeclarativeFilter {
93        predicate: LanguageExpressionDef,
94        steps: Vec<BuilderStep>,
95    },
96    /// Declarative choice/when/otherwise using language predicates, resolved at route-add time.
97    DeclarativeChoice {
98        whens: Vec<DeclarativeWhenStep>,
99        otherwise: Option<Vec<BuilderStep>>,
100    },
101    /// Declarative script step evaluated by language and written to body.
102    DeclarativeScript {
103        expression: LanguageExpressionDef,
104    },
105    DeclarativeFunction {
106        definition: camel_api::FunctionDefinition,
107    },
108    /// Declarative split using a language expression, resolved at route-add time.
109    DeclarativeSplit {
110        expression: LanguageExpressionDef,
111        aggregation: camel_api::splitter::AggregationStrategy,
112        parallel: bool,
113        parallel_limit: Option<usize>,
114        stop_on_exception: bool,
115        steps: Vec<BuilderStep>,
116    },
117    /// Declarative stream split using a streaming split expression, resolved at route-add time.
118    DeclarativeStreamSplit {
119        stream_config: camel_api::StreamSplitConfig,
120        aggregation: camel_api::splitter::AggregationStrategy,
121        stop_on_exception: bool,
122        steps: Vec<BuilderStep>,
123    },
124    DeclarativeDynamicRouter {
125        expression: LanguageExpressionDef,
126        uri_delimiter: String,
127        cache_size: i32,
128        ignore_invalid_endpoints: bool,
129        max_iterations: usize,
130    },
131    DeclarativeRoutingSlip {
132        expression: LanguageExpressionDef,
133        uri_delimiter: String,
134        cache_size: i32,
135        ignore_invalid_endpoints: bool,
136    },
137    /// A Splitter sub-pipeline: config + nested steps to execute per fragment.
138    Split {
139        config: SplitterConfig,
140        steps: Vec<BuilderStep>,
141    },
142    /// An Aggregator step: collects exchanges by correlation key, emits when complete.
143    Aggregate {
144        config: AggregatorConfig,
145    },
146    /// A Filter sub-pipeline: predicate + nested steps executed only when predicate is true.
147    Filter {
148        predicate: FilterPredicate,
149        steps: Vec<BuilderStep>,
150    },
151    /// A Choice step: evaluates when-clauses in order, routes to the first match.
152    /// If no when matches, the optional otherwise branch is used.
153    Choice {
154        whens: Vec<WhenStep>,
155        otherwise: Option<Vec<BuilderStep>>,
156    },
157    /// A WireTap step: sends a clone of the exchange to a tap endpoint (fire-and-forget).
158    WireTap {
159        uri: String,
160    },
161    /// A Multicast step: sends the same exchange to multiple destinations.
162    Multicast {
163        steps: Vec<BuilderStep>,
164        config: MulticastConfig,
165    },
166    /// Declarative log step with a language-evaluated message, resolved at route-add time.
167    DeclarativeLog {
168        level: camel_processor::LogLevel,
169        message: ValueSourceDef,
170    },
171    /// Bean invocation step — resolved at route-add time.
172    Bean {
173        name: String,
174        method: String,
175    },
176    /// Script step: executes a script that can mutate the exchange.
177    /// The script has access to `headers`, `properties`, and `body`.
178    Script {
179        language: String,
180        script: String,
181    },
182    /// Throttle step: rate limiting with configurable behavior when limit exceeded.
183    Throttle {
184        config: camel_api::ThrottlerConfig,
185        steps: Vec<BuilderStep>,
186    },
187    /// LoadBalance step: distributes exchanges across multiple endpoints using a strategy.
188    LoadBalance {
189        config: camel_api::LoadBalancerConfig,
190        steps: Vec<BuilderStep>,
191    },
192    /// DynamicRouter step: routes exchanges dynamically based on expression evaluation.
193    DynamicRouter {
194        config: camel_api::DynamicRouterConfig,
195    },
196    RoutingSlip {
197        config: camel_api::RoutingSlipConfig,
198    },
199    RecipientList {
200        config: camel_api::recipient_list::RecipientListConfig,
201    },
202    DeclarativeRecipientList {
203        expression: LanguageExpressionDef,
204        delimiter: String,
205        parallel: bool,
206        parallel_limit: Option<usize>,
207        stop_on_exception: bool,
208        aggregation: String,
209    },
210    Delay {
211        config: camel_api::DelayConfig,
212    },
213    /// Runtime loop with closure-based predicate (programmatic DSL).
214    Loop {
215        config: LoopConfig,
216        steps: Vec<BuilderStep>,
217    },
218    /// Declarative loop with optional language-based while predicate (YAML DSL).
219    DeclarativeLoop {
220        count: Option<usize>,
221        while_predicate: Option<LanguageExpressionDef>,
222        steps: Vec<BuilderStep>,
223        max_iterations: Option<usize>,
224    },
225    /// EIP-7 enrich: synchronous content enrichment via a resolved producer.
226    Enrich {
227        uri: String,
228        strategy: Option<String>,
229        timeout_ms: Option<u64>,
230    },
231    /// EIP-7 pollEnrich: blocking poll of a PollingConsumer with timeout.
232    PollEnrich {
233        uri: String,
234        strategy: Option<String>,
235        timeout_ms: Option<u64>,
236    },
237    /// Validate step: evaluates a language expression as predicate.
238    /// Exchange passes if predicate returns true; else CamelError::ValidationError.
239    Validate {
240        predicate: LanguageExpressionDef,
241    },
242    /// Claim Check step (EIP). Transforms the exchange body to/from a
243    /// `ClaimCheckRepository` by key. Process-mode, no child pipeline.
244    /// `filter` enables selective merge-back of body/headers during checkout.
245    ClaimCheck {
246        repository: String,
247        operation: String,
248        key: LanguageExpressionDef,
249        filter: Option<String>,
250    },
251    /// Sampling step (EIP). Passes 1 of every N exchanges (counter-based,
252    /// deterministic). Non-sampled exchanges get CamelStop=true (drop semantics).
253    /// Process-mode, stateless. No StepLifecycle — counter is route-scoped.
254    Sampling {
255        period: usize,
256    },
257    /// Sort step (EIP). Orders a body array by extracting a sort key
258    /// from each element via a language expression. Process-mode, stateless.
259    Sort {
260        expression: LanguageExpressionDef,
261        reverse: bool,
262    },
263    /// Idempotent Consumer step (EIP). Wraps a child sub-pipeline that runs
264    /// only when the message-id is NOT present in the named repository.
265    /// Compiled to a `IdempotentConsumerSegment` (OutcomePipeline, segment-mode).
266    IdempotentConsumer {
267        repository: String,
268        expression: LanguageExpressionDef,
269        steps: Vec<BuilderStep>,
270        eager: bool,
271        remove_on_failure: bool,
272    },
273    /// Declarative doTry/doCatch/doFinally, resolved at route-add time.
274    DeclarativeDoTry {
275        try_steps: Vec<BuilderStep>,
276        catch: Vec<DoTryCatchClauseBuilder>,
277        finally: Option<DoTryFinallyBuilder>,
278    },
279    /// Resequencer EIP: resequences exchanges by sequence number.
280    /// Must be a top-level step (not nested inside structural EIPs).
281    Resequence {
282        policy_config: ResequencePolicyConfig,
283    },
284}
285
286/// An unresolved route definition. "to" URIs have not been resolved to producers yet.
287pub struct RouteDefinition {
288    pub(crate) from_uri: String,
289    pub(crate) steps: Vec<BuilderStep>,
290    /// Optional per-route error handler config. Takes precedence over the global one.
291    pub(crate) error_handler: Option<ErrorHandlerConfig>,
292    /// Optional circuit breaker config. Applied between error handler and step pipeline.
293    pub(crate) circuit_breaker: Option<CircuitBreakerConfig>,
294    pub(crate) security_policy: Option<SecurityPolicyConfig>,
295    /// Optional token authenticator for validating JWT/OAuth tokens.
296    pub(crate) security_authenticator: Option<Arc<dyn TokenAuthenticator>>,
297    /// Optional Unit of Work config for in-flight tracking and completion hooks.
298    pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
299    /// User override for the consumer's concurrency model. `None` means
300    /// "use whatever the consumer declares".
301    pub(crate) concurrency: Option<ConcurrencyModel>,
302    /// Unique identifier for this route. Required.
303    pub(crate) route_id: String,
304    /// Whether this route should start automatically when the context starts.
305    pub(crate) auto_startup: bool,
306    /// Order in which routes are started. Lower values start first.
307    pub(crate) startup_order: i32,
308    pub(crate) source_hash: Option<u64>,
309}
310
311impl RouteDefinition {
312    /// Create a new route definition with the required route ID.
313    pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
314        Self {
315            from_uri: from_uri.into(),
316            steps,
317            error_handler: None,
318            circuit_breaker: None,
319            security_policy: None,
320            security_authenticator: None,
321            unit_of_work: None,
322            concurrency: None,
323            route_id: String::new(), // Will be set by with_route_id()
324            auto_startup: true,
325            startup_order: 1000,
326            source_hash: None,
327        }
328    }
329
330    /// The source endpoint URI.
331    pub fn from_uri(&self) -> &str {
332        &self.from_uri
333    }
334
335    /// The steps in this route definition.
336    pub fn steps(&self) -> &[BuilderStep] {
337        &self.steps
338    }
339
340    /// Transform the step list, consuming and returning self.
341    /// Used for post-parse instrumentation (e.g. benchmark timing injection
342    /// around `To` steps). The field stays private; callers rebuild via this
343    /// consuming method rather than holding a `&mut` alias.
344    pub fn map_steps(mut self, f: impl FnOnce(Vec<BuilderStep>) -> Vec<BuilderStep>) -> Self {
345        self.steps = f(self.steps);
346        self
347    }
348
349    /// Set a per-route error handler, overriding the global one.
350    pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
351        self.error_handler = Some(config);
352        self
353    }
354
355    /// Get the route-level error handler config, if set.
356    pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
357        self.error_handler.as_ref()
358    }
359
360    /// Set a circuit breaker for this route.
361    pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
362        self.circuit_breaker = Some(config);
363        self
364    }
365
366    /// Set a security policy for this route.
367    pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
368        self.security_policy = Some(config);
369        self
370    }
371
372    /// Set a token authenticator for this route.
373    pub fn with_security_authenticator(
374        mut self,
375        authenticator: Arc<dyn TokenAuthenticator>,
376    ) -> Self {
377        self.security_authenticator = Some(authenticator);
378        self
379    }
380
381    /// Set a unit of work config for this route.
382    pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
383        self.unit_of_work = Some(config);
384        self
385    }
386
387    /// Get the unit of work config, if set.
388    pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
389        self.unit_of_work.as_ref()
390    }
391
392    /// Get the circuit breaker config, if set.
393    pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
394        self.circuit_breaker.as_ref()
395    }
396
397    pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
398        self.security_policy.as_ref()
399    }
400
401    pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
402        self.security_authenticator.as_ref()
403    }
404
405    /// User-specified concurrency override, if any.
406    pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
407        self.concurrency.as_ref()
408    }
409
410    /// Override the consumer's concurrency model for this route.
411    pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
412        self.concurrency = Some(model);
413        self
414    }
415
416    /// Get the route ID.
417    pub fn route_id(&self) -> &str {
418        &self.route_id
419    }
420
421    /// Whether this route should start automatically when the context starts.
422    pub fn auto_startup(&self) -> bool {
423        self.auto_startup
424    }
425
426    /// Order in which routes are started. Lower values start first.
427    pub fn startup_order(&self) -> i32 {
428        self.startup_order
429    }
430
431    /// Set a unique identifier for this route.
432    pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
433        self.route_id = id.into();
434        self
435    }
436
437    /// Set whether this route should start automatically.
438    pub fn with_auto_startup(mut self, auto: bool) -> Self {
439        self.auto_startup = auto;
440        self
441    }
442
443    /// Set the startup order. Lower values start first.
444    pub fn with_startup_order(mut self, order: i32) -> Self {
445        self.startup_order = order;
446        self
447    }
448
449    pub fn with_source_hash(mut self, hash: u64) -> Self {
450        self.source_hash = Some(hash);
451        self
452    }
453
454    pub fn source_hash(&self) -> Option<u64> {
455        self.source_hash
456    }
457
458    /// Extract the metadata fields needed for introspection.
459    /// This is used by RouteController to store route info without the non-Sync steps.
460    pub fn to_info(&self) -> RouteDefinitionInfo {
461        RouteDefinitionInfo {
462            route_id: self.route_id.clone(),
463            auto_startup: self.auto_startup,
464            startup_order: self.startup_order,
465            source_hash: self.source_hash,
466        }
467    }
468}
469
470/// Minimal route definition metadata for introspection.
471///
472/// This struct contains only the metadata fields from [`RouteDefinition`]
473/// that are needed for route lifecycle management, without the `steps` field
474/// (which contains non-Sync types and cannot be stored in a Sync struct).
475#[derive(Clone)]
476pub struct RouteDefinitionInfo {
477    route_id: String,
478    auto_startup: bool,
479    startup_order: i32,
480    pub(crate) source_hash: Option<u64>,
481}
482
483impl RouteDefinitionInfo {
484    /// Get the route ID.
485    pub fn route_id(&self) -> &str {
486        &self.route_id
487    }
488
489    /// Whether this route should start automatically when the context starts.
490    pub fn auto_startup(&self) -> bool {
491        self.auto_startup
492    }
493
494    /// Order in which routes are started. Lower values start first.
495    pub fn startup_order(&self) -> i32 {
496        self.startup_order
497    }
498
499    pub fn source_hash(&self) -> Option<u64> {
500        self.source_hash
501    }
502}
503
504#[cfg(test)]
505mod tests {
506    use super::*;
507
508    /// Golden debug output for EVERY BuilderStep variant.
509    ///
510    /// Before refactoring the manual `Debug` impl to `#[derive(Debug)]`, this test
511    /// locks the exact output of every variant so the refactor can be verified.
512    #[test]
513    fn golden_debug_output_all_variants() {
514        use camel_api::declarative::LanguageExpressionDef;
515        use camel_api::loop_eip::LoopMode;
516        use camel_api::recipient_list::RecipientListConfig;
517        use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
518        use camel_api::{
519            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
520            FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
521            Value,
522        };
523        use std::sync::Arc;
524
525        let expr = LanguageExpressionDef {
526            language: "simple".into(),
527            source: "${body}".into(),
528        };
529
530        // -- group A: trivial / non-recursive ----------------------------------
531        // derive(Debug) emits the variant without the enum name for unit/tuple variants.
532        assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
533        assert_eq!(
534            format!(
535                "{:?}",
536                BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
537            ),
538            "Processor(BoxProcessor(...))"
539        );
540        assert_eq!(
541            format!("{:?}", BuilderStep::To("mock:out".into())),
542            "To(\"mock:out\")"
543        );
544
545        // -- group B: variant with named fields (no sub-steps) -----------------
546        // derive(Debug) emits ALL fields, not the redacted `..` form.
547        assert_eq!(
548            format!(
549                "{:?}",
550                BuilderStep::Log {
551                    level: camel_processor::LogLevel::Info,
552                    message: "hello".into(),
553                }
554            ),
555            "Log { level: Info, message: \"hello\" }"
556        );
557
558        assert_eq!(
559            format!(
560                "{:?}",
561                BuilderStep::DeclarativeSetHeader {
562                    key: "k".into(),
563                    value: ValueSourceDef::Literal(Value::String("v".into())),
564                }
565            ),
566            "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
567        );
568
569        assert_eq!(
570            format!(
571                "{:?}",
572                BuilderStep::DeclarativeSetHeaderIfAbsent {
573                    key: "k".into(),
574                    value: ValueSourceDef::Literal(Value::String("v".into())),
575                }
576            ),
577            "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
578        );
579
580        assert_eq!(
581            format!(
582                "{:?}",
583                BuilderStep::DeclarativeSetBody {
584                    value: ValueSourceDef::Literal(Value::String("v".into())),
585                }
586            ),
587            "DeclarativeSetBody { value: Literal(String(\"v\")) }"
588        );
589
590        assert_eq!(
591            format!(
592                "{:?}",
593                BuilderStep::DeclarativeSetProperty {
594                    key: "prop".into(),
595                    value_source: ValueSourceDef::Literal(Value::String("v".into())),
596                }
597            ),
598            "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
599        );
600
601        assert_eq!(
602            format!(
603                "{:?}",
604                BuilderStep::DeclarativeScript {
605                    expression: expr.clone(),
606                }
607            ),
608            "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
609        );
610
611        // DeclarativeFunction needs a FunctionDefinition
612        let func_def = FunctionDefinition {
613            id: FunctionId("test-id".into()),
614            runtime: "my_runtime".into(),
615            source: "${body}".into(),
616            timeout_ms: 5000,
617            route_id: None,
618            step_index: None,
619        };
620        assert_eq!(
621            format!(
622                "{:?}",
623                BuilderStep::DeclarativeFunction {
624                    definition: func_def,
625                }
626            ),
627            "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
628        );
629
630        assert_eq!(
631            format!(
632                "{:?}",
633                BuilderStep::WireTap {
634                    uri: "mock:tap".into(),
635                }
636            ),
637            "WireTap { uri: \"mock:tap\" }"
638        );
639
640        assert_eq!(
641            format!(
642                "{:?}",
643                BuilderStep::DeclarativeLog {
644                    level: camel_processor::LogLevel::Info,
645                    message: ValueSourceDef::Expression(expr.clone()),
646                }
647            ),
648            "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
649        );
650
651        assert_eq!(
652            format!(
653                "{:?}",
654                BuilderStep::Bean {
655                    name: "myBean".into(),
656                    method: "process".into(),
657                }
658            ),
659            "Bean { name: \"myBean\", method: \"process\" }"
660        );
661
662        assert_eq!(
663            format!(
664                "{:?}",
665                BuilderStep::Script {
666                    language: "js".into(),
667                    script: "body".into(),
668                }
669            ),
670            "Script { language: \"js\", script: \"body\" }"
671        );
672
673        assert_eq!(
674            format!(
675                "{:?}",
676                BuilderStep::Aggregate {
677                    config: camel_api::AggregatorConfig::correlate_by("id")
678                        .complete_when_size(1)
679                        .build()
680                        .unwrap(),
681                }
682            ),
683            "Aggregate { config: AggregatorConfig { header_name: \"id\", completion: Single(Size(1)), correlation: HeaderName(\"id\"), strategy: CollectAll, max_buckets: Some(10000), bucket_ttl: Some(300s), force_completion_on_stop: false, discard_on_timeout: false, max_timeout_tasks: 1024 } }"
684        );
685
686        assert_eq!(
687            format!(
688                "{:?}",
689                BuilderStep::DynamicRouter {
690                    config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
691                        "mock:dr".into()
692                    ))),
693                }
694            ),
695            "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
696        );
697
698        assert_eq!(
699            format!(
700                "{:?}",
701                BuilderStep::RoutingSlip {
702                    config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
703                }
704            ),
705            "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
706        );
707
708        assert_eq!(
709            format!(
710                "{:?}",
711                BuilderStep::RecipientList {
712                    config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
713                }
714            ),
715            "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
716        );
717
718        assert_eq!(
719            format!(
720                "{:?}",
721                BuilderStep::Enrich {
722                    uri: "mock:enrich".into(),
723                    strategy: Some("agg".into()),
724                    timeout_ms: Some(1000),
725                }
726            ),
727            "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
728        );
729
730        assert_eq!(
731            format!(
732                "{:?}",
733                BuilderStep::PollEnrich {
734                    uri: "mock:poll".into(),
735                    strategy: None,
736                    timeout_ms: None,
737                }
738            ),
739            "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
740        );
741
742        assert_eq!(
743            format!(
744                "{:?}",
745                BuilderStep::Validate {
746                    predicate: expr.clone(),
747                }
748            ),
749            "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
750        );
751
752        assert_eq!(
753            format!("{:?}", BuilderStep::Sampling { period: 100 }),
754            "Sampling { period: 100 }"
755        );
756
757        assert_eq!(
758            format!(
759                "{:?}",
760                BuilderStep::Resequence {
761                    policy_config: Default::default(),
762                }
763            ),
764            "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
765        );
766
767        // -- group C: variants with named fields (sub-steps is Vec) ------------
768        assert_eq!(
769            format!(
770                "{:?}",
771                BuilderStep::DeclarativeFilter {
772                    predicate: expr.clone(),
773                    steps: vec![BuilderStep::Stop],
774                }
775            ),
776            "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
777        );
778
779        assert_eq!(
780            format!(
781                "{:?}",
782                BuilderStep::DeclarativeSplit {
783                    expression: expr.clone(),
784                    aggregation: AggregationStrategy::Original,
785                    parallel: false,
786                    parallel_limit: Some(2),
787                    stop_on_exception: true,
788                    steps: vec![BuilderStep::Stop],
789                }
790            ),
791            "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
792        );
793
794        assert_eq!(
795            format!(
796                "{:?}",
797                BuilderStep::Split {
798                    config: camel_api::splitter::SplitterConfig::new(
799                        camel_api::splitter::split_body_lines()
800                    ),
801                    steps: vec![BuilderStep::Stop],
802                }
803            ),
804            "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
805        );
806
807        assert_eq!(
808            format!(
809                "{:?}",
810                BuilderStep::Filter {
811                    predicate: FilterPredicate::new(|_: &Exchange| true),
812                    steps: vec![BuilderStep::Stop],
813                }
814            ),
815            "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
816        );
817
818        assert_eq!(
819            format!(
820                "{:?}",
821                BuilderStep::Throttle {
822                    config: camel_api::ThrottlerConfig::new(
823                        10,
824                        std::time::Duration::from_millis(10)
825                    ),
826                    steps: vec![BuilderStep::Stop],
827                }
828            ),
829            "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
830        );
831
832        assert_eq!(
833            format!(
834                "{:?}",
835                BuilderStep::LoadBalance {
836                    config: camel_api::LoadBalancerConfig::round_robin(),
837                    steps: vec![BuilderStep::To("mock:l1".into())],
838                }
839            ),
840            "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
841        );
842
843        assert_eq!(
844            format!(
845                "{:?}",
846                BuilderStep::Delay {
847                    config: camel_api::DelayConfig::new(500),
848                }
849            ),
850            "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
851        );
852
853        // -- group D: Choice / DeclarativeChoice --------------------------------
854        // derive(Debug) emits full field enumeration; WhenStep Debug is a
855        // full struct listing; nested BuilderStep::Stop stays as `Stop`.
856        assert_eq!(
857            format!(
858                "{:?}",
859                BuilderStep::Choice {
860                    whens: vec![WhenStep {
861                        predicate: FilterPredicate::new(|_: &Exchange| true),
862                        steps: vec![BuilderStep::To("mock:a".into())],
863                    }],
864                    otherwise: None,
865                }
866            ),
867            "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
868        );
869
870        assert_eq!(
871            format!(
872                "{:?}",
873                BuilderStep::DeclarativeChoice {
874                    whens: vec![DeclarativeWhenStep {
875                        predicate: expr.clone(),
876                        steps: vec![BuilderStep::Stop],
877                    }],
878                    otherwise: Some(vec![BuilderStep::Stop]),
879                }
880            ),
881            "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
882        );
883
884        // -- group E: Multicast -------------------------------------------------
885        assert_eq!(
886            format!(
887                "{:?}",
888                BuilderStep::Multicast {
889                    steps: vec![BuilderStep::To("direct:a".into())],
890                    config: MulticastConfig::new(),
891                }
892            ),
893            "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
894        );
895
896        // -- group F: DeclarativeDynamicRouter / DeclarativeRoutingSlip --------
897        assert_eq!(
898            format!(
899                "{:?}",
900                BuilderStep::DeclarativeDynamicRouter {
901                    expression: expr.clone(),
902                    uri_delimiter: ",".into(),
903                    cache_size: 1000,
904                    ignore_invalid_endpoints: false,
905                    max_iterations: 1000,
906                }
907            ),
908            "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
909        );
910
911        assert_eq!(
912            format!(
913                "{:?}",
914                BuilderStep::DeclarativeRoutingSlip {
915                    expression: expr.clone(),
916                    uri_delimiter: ",".into(),
917                    cache_size: 1000,
918                    ignore_invalid_endpoints: false,
919                }
920            ),
921            "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
922        );
923
924        // -- group G: DeclarativeRecipientList ---------------------------------
925        assert_eq!(
926            format!(
927                "{:?}",
928                BuilderStep::DeclarativeRecipientList {
929                    expression: expr.clone(),
930                    delimiter: ",".into(),
931                    parallel: false,
932                    parallel_limit: None,
933                    stop_on_exception: false,
934                    aggregation: "original".into(),
935                }
936            ),
937            "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
938        );
939
940        // -- group H: Loop / DeclarativeLoop -----------------------------------
941        assert_eq!(
942            format!(
943                "{:?}",
944                BuilderStep::Loop {
945                    config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
946                    steps: vec![],
947                }
948            ),
949            "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
950        );
951
952        assert_eq!(
953            format!(
954                "{:?}",
955                BuilderStep::DeclarativeLoop {
956                    count: Some(5),
957                    while_predicate: None,
958                    steps: vec![],
959                    max_iterations: Some(100),
960                }
961            ),
962            "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
963        );
964
965        // -- group I: ClaimCheck -----------------------------------------------
966        assert_eq!(
967            format!(
968                "{:?}",
969                BuilderStep::ClaimCheck {
970                    repository: "myRepo".into(),
971                    operation: "checkout".into(),
972                    key: expr.clone(),
973                    filter: None,
974                }
975            ),
976            "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
977        );
978
979        // -- group J: Sort -----------------------------------------------------
980        assert_eq!(
981            format!(
982                "{:?}",
983                BuilderStep::Sort {
984                    expression: expr.clone(),
985                    reverse: false,
986                }
987            ),
988            "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
989        );
990
991        // -- group K: IdempotentConsumer ---------------------------------------
992        assert_eq!(
993            format!(
994                "{:?}",
995                BuilderStep::IdempotentConsumer {
996                    repository: "myRepo".into(),
997                    expression: expr.clone(),
998                    steps: vec![],
999                    eager: true,
1000                    remove_on_failure: false,
1001                }
1002            ),
1003            "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
1004        );
1005
1006        // -- group L: DeclarativeDoTry -----------------------------------------
1007        assert_eq!(
1008            format!(
1009                "{:?}",
1010                BuilderStep::DeclarativeDoTry {
1011                    try_steps: vec![BuilderStep::Stop],
1012                    catch: vec![],
1013                    finally: None,
1014                }
1015            ),
1016            "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1017        );
1018
1019        // -- group M: DeclarativeStreamSplit -----------------------------------
1020        assert_eq!(
1021            format!(
1022                "{:?}",
1023                BuilderStep::DeclarativeStreamSplit {
1024                    stream_config: StreamSplitConfig {
1025                        format: StreamSplitFormat::Ndjson,
1026                        max_record_bytes: 1024 * 1024,
1027                        batch_size: 1,
1028                        chunk_size: None,
1029                        include_origin: true,
1030                    },
1031                    aggregation: AggregationStrategy::Original,
1032                    stop_on_exception: true,
1033                    steps: vec![BuilderStep::Stop],
1034                }
1035            ),
1036            "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] }"
1037        );
1038    }
1039
1040    #[test]
1041    fn test_builder_step_multicast_variant() {
1042        use camel_api::MulticastConfig;
1043
1044        let step = BuilderStep::Multicast {
1045            steps: vec![BuilderStep::To("direct:a".into())],
1046            config: MulticastConfig::new(),
1047        };
1048
1049        assert!(matches!(step, BuilderStep::Multicast { .. }));
1050    }
1051
1052    #[test]
1053    fn test_route_definition_defaults() {
1054        let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1055        assert_eq!(def.route_id(), "test-route");
1056        assert!(def.auto_startup());
1057        assert_eq!(def.startup_order(), 1000);
1058    }
1059
1060    #[test]
1061    fn test_route_definition_builders() {
1062        let def = RouteDefinition::new("direct:test", vec![])
1063            .with_route_id("my-route")
1064            .with_auto_startup(false)
1065            .with_startup_order(50);
1066        assert_eq!(def.route_id(), "my-route");
1067        assert!(!def.auto_startup());
1068        assert_eq!(def.startup_order(), 50);
1069    }
1070
1071    #[test]
1072    fn test_route_definition_accessors_cover_core_fields() {
1073        let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1074            .with_route_id("accessor-route");
1075
1076        assert_eq!(def.from_uri(), "direct:in");
1077        assert_eq!(def.steps().len(), 1);
1078        assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1079    }
1080
1081    #[test]
1082    fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1083        use camel_api::circuit_breaker::CircuitBreakerConfig;
1084        use camel_api::error_handler::ErrorHandlerConfig;
1085        use camel_component_api::ConcurrencyModel;
1086
1087        let def = RouteDefinition::new("direct:test", vec![])
1088            .with_route_id("eh-route")
1089            .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1090            .with_circuit_breaker(CircuitBreakerConfig::new())
1091            .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1092
1093        let eh = def
1094            .error_handler_config()
1095            .expect("error handler should be set");
1096        assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1097        assert!(def.circuit_breaker_config().is_some());
1098        assert!(matches!(
1099            def.concurrency_override(),
1100            Some(ConcurrencyModel::Concurrent { max: Some(4) })
1101        ));
1102    }
1103
1104    #[test]
1105    fn test_builder_step_debug_covers_many_variants() {
1106        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1107        use camel_api::{
1108            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1109            OpaqueProcessor, RoutingSlipConfig, Value,
1110        };
1111        use std::sync::Arc;
1112
1113        let expr = LanguageExpressionDef {
1114            language: "simple".into(),
1115            source: "${body}".into(),
1116        };
1117
1118        let steps = vec![
1119            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1120            BuilderStep::To("mock:out".into()),
1121            BuilderStep::Stop,
1122            BuilderStep::Log {
1123                level: camel_processor::LogLevel::Info,
1124                message: "hello".into(),
1125            },
1126            BuilderStep::DeclarativeSetHeader {
1127                key: "k".into(),
1128                value: ValueSourceDef::Literal(Value::String("v".into())),
1129            },
1130            BuilderStep::DeclarativeSetBody {
1131                value: ValueSourceDef::Expression(expr.clone()),
1132            },
1133            BuilderStep::DeclarativeFilter {
1134                predicate: expr.clone(),
1135                steps: vec![BuilderStep::Stop],
1136            },
1137            BuilderStep::DeclarativeChoice {
1138                whens: vec![DeclarativeWhenStep {
1139                    predicate: expr.clone(),
1140                    steps: vec![BuilderStep::Stop],
1141                }],
1142                otherwise: Some(vec![BuilderStep::Stop]),
1143            },
1144            BuilderStep::DeclarativeScript {
1145                expression: expr.clone(),
1146            },
1147            BuilderStep::DeclarativeSplit {
1148                expression: expr.clone(),
1149                aggregation: AggregationStrategy::Original,
1150                parallel: false,
1151                parallel_limit: Some(2),
1152                stop_on_exception: true,
1153                steps: vec![BuilderStep::Stop],
1154            },
1155            BuilderStep::Split {
1156                config: SplitterConfig::new(split_body_lines()),
1157                steps: vec![BuilderStep::Stop],
1158            },
1159            BuilderStep::Aggregate {
1160                config: camel_api::AggregatorConfig::correlate_by("id")
1161                    .complete_when_size(1)
1162                    .build()
1163                    .unwrap(),
1164            },
1165            BuilderStep::Filter {
1166                predicate: FilterPredicate::new(|_: &Exchange| true),
1167                steps: vec![BuilderStep::Stop],
1168            },
1169            BuilderStep::WireTap {
1170                uri: "mock:tap".into(),
1171            },
1172            BuilderStep::DeclarativeLog {
1173                level: camel_processor::LogLevel::Info,
1174                message: ValueSourceDef::Expression(expr.clone()),
1175            },
1176            BuilderStep::Bean {
1177                name: "bean".into(),
1178                method: "call".into(),
1179            },
1180            BuilderStep::Script {
1181                language: "rhai".into(),
1182                script: "body".into(),
1183            },
1184            BuilderStep::Throttle {
1185                config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1186                steps: vec![BuilderStep::Stop],
1187            },
1188            BuilderStep::LoadBalance {
1189                config: camel_api::LoadBalancerConfig::round_robin(),
1190                steps: vec![BuilderStep::To("mock:l1".into())],
1191            },
1192            BuilderStep::DynamicRouter {
1193                config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1194            },
1195            BuilderStep::RoutingSlip {
1196                config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1197            },
1198        ];
1199
1200        for step in steps {
1201            let dbg = format!("{step:?}");
1202            assert!(!dbg.is_empty());
1203        }
1204    }
1205
1206    #[test]
1207    fn test_route_definition_to_info_preserves_metadata() {
1208        let info = RouteDefinition::new("direct:test", vec![])
1209            .with_route_id("meta-route")
1210            .with_auto_startup(false)
1211            .with_startup_order(7)
1212            .to_info();
1213
1214        assert_eq!(info.route_id(), "meta-route");
1215        assert!(!info.auto_startup());
1216        assert_eq!(info.startup_order(), 7);
1217    }
1218
1219    #[test]
1220    fn test_choice_builder_step_debug() {
1221        use camel_api::FilterPredicate;
1222
1223        fn always_true(_: &camel_api::Exchange) -> bool {
1224            true
1225        }
1226
1227        let step = BuilderStep::Choice {
1228            whens: vec![WhenStep {
1229                predicate: FilterPredicate::new(always_true),
1230                steps: vec![BuilderStep::To("mock:a".into())],
1231            }],
1232            otherwise: None,
1233        };
1234        let debug = format!("{step:?}");
1235        assert!(debug.contains("Choice"));
1236    }
1237
1238    #[test]
1239    fn test_route_definition_unit_of_work() {
1240        use camel_api::UnitOfWorkConfig;
1241        let config = UnitOfWorkConfig {
1242            on_complete: Some("log:complete".into()),
1243            on_failure: Some("log:failed".into()),
1244        };
1245        let def = RouteDefinition::new("direct:test", vec![])
1246            .with_route_id("uow-test")
1247            .with_unit_of_work(config.clone());
1248        assert_eq!(
1249            def.unit_of_work_config().unwrap().on_complete.as_deref(),
1250            Some("log:complete")
1251        );
1252        assert_eq!(
1253            def.unit_of_work_config().unwrap().on_failure.as_deref(),
1254            Some("log:failed")
1255        );
1256
1257        let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1258        assert!(def_no_uow.unit_of_work_config().is_none());
1259    }
1260
1261    #[test]
1262    fn test_route_definition_security_policy_accessor() {
1263        use async_trait::async_trait;
1264        use camel_api::CamelError;
1265        use camel_api::Exchange;
1266        use camel_api::security_policy::{
1267            AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1268        };
1269
1270        struct StubPolicy;
1271        #[async_trait]
1272        impl SecurityPolicy for StubPolicy {
1273            async fn evaluate(
1274                &self,
1275                _exchange: &mut Exchange,
1276            ) -> Result<AuthorizationDecision, CamelError> {
1277                Ok(AuthorizationDecision::Granted {
1278                    principal: Principal {
1279                        subject: "test".into(),
1280                        issuer: "test".into(),
1281                        audience: vec![],
1282                        scopes: vec![],
1283                        roles: vec![],
1284                        claims: serde_json::Value::Null,
1285                    },
1286                })
1287            }
1288        }
1289
1290        let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1291        assert!(def_no_sp.security_policy_config().is_none());
1292
1293        let def = RouteDefinition::new("direct:test", vec![])
1294            .with_route_id("sp-test")
1295            .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1296        assert!(def.security_policy_config().is_some());
1297    }
1298
1299    #[test]
1300    fn test_route_definition_security_authenticator_accessor() {
1301        use camel_api::security_policy::Principal;
1302
1303        struct TestAuth;
1304        #[async_trait::async_trait]
1305        impl TokenAuthenticator for TestAuth {
1306            async fn authenticate_bearer(
1307                &self,
1308                _token: &str,
1309            ) -> Result<Principal, camel_api::CamelError> {
1310                Ok(Principal {
1311                    subject: "test".into(),
1312                    issuer: "test".into(),
1313                    audience: vec![],
1314                    scopes: vec![],
1315                    roles: vec![],
1316                    claims: serde_json::Value::Null,
1317                })
1318            }
1319        }
1320
1321        let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1322        assert!(def_no_auth.security_authenticator().is_none());
1323
1324        let auth = Arc::new(TestAuth);
1325        let def = RouteDefinition::new("direct:test".to_string(), vec![])
1326            .with_security_authenticator(auth);
1327        assert!(def.security_authenticator().is_some());
1328    }
1329
1330    #[test]
1331    fn test_map_steps_swaps_steps_and_preserves_other_fields() {
1332        let original = RouteDefinition::new(
1333            "direct:test".to_string(),
1334            vec![BuilderStep::To("mock:a".into()), BuilderStep::Stop],
1335        )
1336        .with_route_id("my-route");
1337
1338        let mapped = original.map_steps(|steps| {
1339            let mut out = Vec::with_capacity(steps.len() + 1);
1340            out.push(BuilderStep::To("mock:prefix".into()));
1341            out.extend(steps);
1342            out
1343        });
1344
1345        // Steps were transformed.
1346        assert_eq!(mapped.steps().len(), 3);
1347        assert!(matches!(mapped.steps()[0], BuilderStep::To(ref s) if s == "mock:prefix"));
1348        assert!(matches!(mapped.steps()[1], BuilderStep::To(ref s) if s == "mock:a"));
1349        // Other fields preserved.
1350        assert_eq!(mapped.route_id(), "my-route");
1351    }
1352}
camel_core/lifecycle/application/route_definition.rs

camel_core/lifecycle/application/
route_definition.rs
