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    /// Set a per-route error handler, overriding the global one.
341    pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
342        self.error_handler = Some(config);
343        self
344    }
345
346    /// Get the route-level error handler config, if set.
347    pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
348        self.error_handler.as_ref()
349    }
350
351    /// Set a circuit breaker for this route.
352    pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
353        self.circuit_breaker = Some(config);
354        self
355    }
356
357    /// Set a security policy for this route.
358    pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
359        self.security_policy = Some(config);
360        self
361    }
362
363    /// Set a token authenticator for this route.
364    pub fn with_security_authenticator(
365        mut self,
366        authenticator: Arc<dyn TokenAuthenticator>,
367    ) -> Self {
368        self.security_authenticator = Some(authenticator);
369        self
370    }
371
372    /// Set a unit of work config for this route.
373    pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
374        self.unit_of_work = Some(config);
375        self
376    }
377
378    /// Get the unit of work config, if set.
379    pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
380        self.unit_of_work.as_ref()
381    }
382
383    /// Get the circuit breaker config, if set.
384    pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
385        self.circuit_breaker.as_ref()
386    }
387
388    pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
389        self.security_policy.as_ref()
390    }
391
392    pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
393        self.security_authenticator.as_ref()
394    }
395
396    /// User-specified concurrency override, if any.
397    pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
398        self.concurrency.as_ref()
399    }
400
401    /// Override the consumer's concurrency model for this route.
402    pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
403        self.concurrency = Some(model);
404        self
405    }
406
407    /// Get the route ID.
408    pub fn route_id(&self) -> &str {
409        &self.route_id
410    }
411
412    /// Whether this route should start automatically when the context starts.
413    pub fn auto_startup(&self) -> bool {
414        self.auto_startup
415    }
416
417    /// Order in which routes are started. Lower values start first.
418    pub fn startup_order(&self) -> i32 {
419        self.startup_order
420    }
421
422    /// Set a unique identifier for this route.
423    pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
424        self.route_id = id.into();
425        self
426    }
427
428    /// Set whether this route should start automatically.
429    pub fn with_auto_startup(mut self, auto: bool) -> Self {
430        self.auto_startup = auto;
431        self
432    }
433
434    /// Set the startup order. Lower values start first.
435    pub fn with_startup_order(mut self, order: i32) -> Self {
436        self.startup_order = order;
437        self
438    }
439
440    pub fn with_source_hash(mut self, hash: u64) -> Self {
441        self.source_hash = Some(hash);
442        self
443    }
444
445    pub fn source_hash(&self) -> Option<u64> {
446        self.source_hash
447    }
448
449    /// Extract the metadata fields needed for introspection.
450    /// This is used by RouteController to store route info without the non-Sync steps.
451    pub fn to_info(&self) -> RouteDefinitionInfo {
452        RouteDefinitionInfo {
453            route_id: self.route_id.clone(),
454            auto_startup: self.auto_startup,
455            startup_order: self.startup_order,
456            source_hash: self.source_hash,
457        }
458    }
459}
460
461/// Minimal route definition metadata for introspection.
462///
463/// This struct contains only the metadata fields from [`RouteDefinition`]
464/// that are needed for route lifecycle management, without the `steps` field
465/// (which contains non-Sync types and cannot be stored in a Sync struct).
466#[derive(Clone)]
467pub struct RouteDefinitionInfo {
468    route_id: String,
469    auto_startup: bool,
470    startup_order: i32,
471    pub(crate) source_hash: Option<u64>,
472}
473
474impl RouteDefinitionInfo {
475    /// Get the route ID.
476    pub fn route_id(&self) -> &str {
477        &self.route_id
478    }
479
480    /// Whether this route should start automatically when the context starts.
481    pub fn auto_startup(&self) -> bool {
482        self.auto_startup
483    }
484
485    /// Order in which routes are started. Lower values start first.
486    pub fn startup_order(&self) -> i32 {
487        self.startup_order
488    }
489
490    pub fn source_hash(&self) -> Option<u64> {
491        self.source_hash
492    }
493}
494
495#[cfg(test)]
496mod tests {
497    use super::*;
498
499    /// Golden debug output for EVERY BuilderStep variant.
500    ///
501    /// Before refactoring the manual `Debug` impl to `#[derive(Debug)]`, this test
502    /// locks the exact output of every variant so the refactor can be verified.
503    #[test]
504    fn golden_debug_output_all_variants() {
505        use camel_api::declarative::LanguageExpressionDef;
506        use camel_api::loop_eip::LoopMode;
507        use camel_api::recipient_list::RecipientListConfig;
508        use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
509        use camel_api::{
510            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
511            FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
512            Value,
513        };
514        use std::sync::Arc;
515
516        let expr = LanguageExpressionDef {
517            language: "simple".into(),
518            source: "${body}".into(),
519        };
520
521        // -- group A: trivial / non-recursive ----------------------------------
522        // derive(Debug) emits the variant without the enum name for unit/tuple variants.
523        assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
524        assert_eq!(
525            format!(
526                "{:?}",
527                BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
528            ),
529            "Processor(BoxProcessor(...))"
530        );
531        assert_eq!(
532            format!("{:?}", BuilderStep::To("mock:out".into())),
533            "To(\"mock:out\")"
534        );
535
536        // -- group B: variant with named fields (no sub-steps) -----------------
537        // derive(Debug) emits ALL fields, not the redacted `..` form.
538        assert_eq!(
539            format!(
540                "{:?}",
541                BuilderStep::Log {
542                    level: camel_processor::LogLevel::Info,
543                    message: "hello".into(),
544                }
545            ),
546            "Log { level: Info, message: \"hello\" }"
547        );
548
549        assert_eq!(
550            format!(
551                "{:?}",
552                BuilderStep::DeclarativeSetHeader {
553                    key: "k".into(),
554                    value: ValueSourceDef::Literal(Value::String("v".into())),
555                }
556            ),
557            "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
558        );
559
560        assert_eq!(
561            format!(
562                "{:?}",
563                BuilderStep::DeclarativeSetHeaderIfAbsent {
564                    key: "k".into(),
565                    value: ValueSourceDef::Literal(Value::String("v".into())),
566                }
567            ),
568            "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
569        );
570
571        assert_eq!(
572            format!(
573                "{:?}",
574                BuilderStep::DeclarativeSetBody {
575                    value: ValueSourceDef::Literal(Value::String("v".into())),
576                }
577            ),
578            "DeclarativeSetBody { value: Literal(String(\"v\")) }"
579        );
580
581        assert_eq!(
582            format!(
583                "{:?}",
584                BuilderStep::DeclarativeSetProperty {
585                    key: "prop".into(),
586                    value_source: ValueSourceDef::Literal(Value::String("v".into())),
587                }
588            ),
589            "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
590        );
591
592        assert_eq!(
593            format!(
594                "{:?}",
595                BuilderStep::DeclarativeScript {
596                    expression: expr.clone(),
597                }
598            ),
599            "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
600        );
601
602        // DeclarativeFunction needs a FunctionDefinition
603        let func_def = FunctionDefinition {
604            id: FunctionId("test-id".into()),
605            runtime: "my_runtime".into(),
606            source: "${body}".into(),
607            timeout_ms: 5000,
608            route_id: None,
609            step_index: None,
610        };
611        assert_eq!(
612            format!(
613                "{:?}",
614                BuilderStep::DeclarativeFunction {
615                    definition: func_def,
616                }
617            ),
618            "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
619        );
620
621        assert_eq!(
622            format!(
623                "{:?}",
624                BuilderStep::WireTap {
625                    uri: "mock:tap".into(),
626                }
627            ),
628            "WireTap { uri: \"mock:tap\" }"
629        );
630
631        assert_eq!(
632            format!(
633                "{:?}",
634                BuilderStep::DeclarativeLog {
635                    level: camel_processor::LogLevel::Info,
636                    message: ValueSourceDef::Expression(expr.clone()),
637                }
638            ),
639            "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
640        );
641
642        assert_eq!(
643            format!(
644                "{:?}",
645                BuilderStep::Bean {
646                    name: "myBean".into(),
647                    method: "process".into(),
648                }
649            ),
650            "Bean { name: \"myBean\", method: \"process\" }"
651        );
652
653        assert_eq!(
654            format!(
655                "{:?}",
656                BuilderStep::Script {
657                    language: "js".into(),
658                    script: "body".into(),
659                }
660            ),
661            "Script { language: \"js\", script: \"body\" }"
662        );
663
664        assert_eq!(
665            format!(
666                "{:?}",
667                BuilderStep::Aggregate {
668                    config: camel_api::AggregatorConfig::correlate_by("id")
669                        .complete_when_size(1)
670                        .build()
671                        .unwrap(),
672                }
673            ),
674            "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 } }"
675        );
676
677        assert_eq!(
678            format!(
679                "{:?}",
680                BuilderStep::DynamicRouter {
681                    config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
682                        "mock:dr".into()
683                    ))),
684                }
685            ),
686            "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
687        );
688
689        assert_eq!(
690            format!(
691                "{:?}",
692                BuilderStep::RoutingSlip {
693                    config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
694                }
695            ),
696            "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
697        );
698
699        assert_eq!(
700            format!(
701                "{:?}",
702                BuilderStep::RecipientList {
703                    config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
704                }
705            ),
706            "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
707        );
708
709        assert_eq!(
710            format!(
711                "{:?}",
712                BuilderStep::Enrich {
713                    uri: "mock:enrich".into(),
714                    strategy: Some("agg".into()),
715                    timeout_ms: Some(1000),
716                }
717            ),
718            "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
719        );
720
721        assert_eq!(
722            format!(
723                "{:?}",
724                BuilderStep::PollEnrich {
725                    uri: "mock:poll".into(),
726                    strategy: None,
727                    timeout_ms: None,
728                }
729            ),
730            "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
731        );
732
733        assert_eq!(
734            format!(
735                "{:?}",
736                BuilderStep::Validate {
737                    predicate: expr.clone(),
738                }
739            ),
740            "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
741        );
742
743        assert_eq!(
744            format!("{:?}", BuilderStep::Sampling { period: 100 }),
745            "Sampling { period: 100 }"
746        );
747
748        assert_eq!(
749            format!(
750                "{:?}",
751                BuilderStep::Resequence {
752                    policy_config: Default::default(),
753                }
754            ),
755            "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
756        );
757
758        // -- group C: variants with named fields (sub-steps is Vec) ------------
759        assert_eq!(
760            format!(
761                "{:?}",
762                BuilderStep::DeclarativeFilter {
763                    predicate: expr.clone(),
764                    steps: vec![BuilderStep::Stop],
765                }
766            ),
767            "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
768        );
769
770        assert_eq!(
771            format!(
772                "{:?}",
773                BuilderStep::DeclarativeSplit {
774                    expression: expr.clone(),
775                    aggregation: AggregationStrategy::Original,
776                    parallel: false,
777                    parallel_limit: Some(2),
778                    stop_on_exception: true,
779                    steps: vec![BuilderStep::Stop],
780                }
781            ),
782            "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
783        );
784
785        assert_eq!(
786            format!(
787                "{:?}",
788                BuilderStep::Split {
789                    config: camel_api::splitter::SplitterConfig::new(
790                        camel_api::splitter::split_body_lines()
791                    ),
792                    steps: vec![BuilderStep::Stop],
793                }
794            ),
795            "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
796        );
797
798        assert_eq!(
799            format!(
800                "{:?}",
801                BuilderStep::Filter {
802                    predicate: FilterPredicate::new(|_: &Exchange| true),
803                    steps: vec![BuilderStep::Stop],
804                }
805            ),
806            "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
807        );
808
809        assert_eq!(
810            format!(
811                "{:?}",
812                BuilderStep::Throttle {
813                    config: camel_api::ThrottlerConfig::new(
814                        10,
815                        std::time::Duration::from_millis(10)
816                    ),
817                    steps: vec![BuilderStep::Stop],
818                }
819            ),
820            "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
821        );
822
823        assert_eq!(
824            format!(
825                "{:?}",
826                BuilderStep::LoadBalance {
827                    config: camel_api::LoadBalancerConfig::round_robin(),
828                    steps: vec![BuilderStep::To("mock:l1".into())],
829                }
830            ),
831            "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
832        );
833
834        assert_eq!(
835            format!(
836                "{:?}",
837                BuilderStep::Delay {
838                    config: camel_api::DelayConfig::new(500),
839                }
840            ),
841            "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
842        );
843
844        // -- group D: Choice / DeclarativeChoice --------------------------------
845        // derive(Debug) emits full field enumeration; WhenStep Debug is a
846        // full struct listing; nested BuilderStep::Stop stays as `Stop`.
847        assert_eq!(
848            format!(
849                "{:?}",
850                BuilderStep::Choice {
851                    whens: vec![WhenStep {
852                        predicate: FilterPredicate::new(|_: &Exchange| true),
853                        steps: vec![BuilderStep::To("mock:a".into())],
854                    }],
855                    otherwise: None,
856                }
857            ),
858            "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
859        );
860
861        assert_eq!(
862            format!(
863                "{:?}",
864                BuilderStep::DeclarativeChoice {
865                    whens: vec![DeclarativeWhenStep {
866                        predicate: expr.clone(),
867                        steps: vec![BuilderStep::Stop],
868                    }],
869                    otherwise: Some(vec![BuilderStep::Stop]),
870                }
871            ),
872            "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
873        );
874
875        // -- group E: Multicast -------------------------------------------------
876        assert_eq!(
877            format!(
878                "{:?}",
879                BuilderStep::Multicast {
880                    steps: vec![BuilderStep::To("direct:a".into())],
881                    config: MulticastConfig::new(),
882                }
883            ),
884            "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
885        );
886
887        // -- group F: DeclarativeDynamicRouter / DeclarativeRoutingSlip --------
888        assert_eq!(
889            format!(
890                "{:?}",
891                BuilderStep::DeclarativeDynamicRouter {
892                    expression: expr.clone(),
893                    uri_delimiter: ",".into(),
894                    cache_size: 1000,
895                    ignore_invalid_endpoints: false,
896                    max_iterations: 1000,
897                }
898            ),
899            "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
900        );
901
902        assert_eq!(
903            format!(
904                "{:?}",
905                BuilderStep::DeclarativeRoutingSlip {
906                    expression: expr.clone(),
907                    uri_delimiter: ",".into(),
908                    cache_size: 1000,
909                    ignore_invalid_endpoints: false,
910                }
911            ),
912            "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
913        );
914
915        // -- group G: DeclarativeRecipientList ---------------------------------
916        assert_eq!(
917            format!(
918                "{:?}",
919                BuilderStep::DeclarativeRecipientList {
920                    expression: expr.clone(),
921                    delimiter: ",".into(),
922                    parallel: false,
923                    parallel_limit: None,
924                    stop_on_exception: false,
925                    aggregation: "original".into(),
926                }
927            ),
928            "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
929        );
930
931        // -- group H: Loop / DeclarativeLoop -----------------------------------
932        assert_eq!(
933            format!(
934                "{:?}",
935                BuilderStep::Loop {
936                    config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
937                    steps: vec![],
938                }
939            ),
940            "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
941        );
942
943        assert_eq!(
944            format!(
945                "{:?}",
946                BuilderStep::DeclarativeLoop {
947                    count: Some(5),
948                    while_predicate: None,
949                    steps: vec![],
950                    max_iterations: Some(100),
951                }
952            ),
953            "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
954        );
955
956        // -- group I: ClaimCheck -----------------------------------------------
957        assert_eq!(
958            format!(
959                "{:?}",
960                BuilderStep::ClaimCheck {
961                    repository: "myRepo".into(),
962                    operation: "checkout".into(),
963                    key: expr.clone(),
964                    filter: None,
965                }
966            ),
967            "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
968        );
969
970        // -- group J: Sort -----------------------------------------------------
971        assert_eq!(
972            format!(
973                "{:?}",
974                BuilderStep::Sort {
975                    expression: expr.clone(),
976                    reverse: false,
977                }
978            ),
979            "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
980        );
981
982        // -- group K: IdempotentConsumer ---------------------------------------
983        assert_eq!(
984            format!(
985                "{:?}",
986                BuilderStep::IdempotentConsumer {
987                    repository: "myRepo".into(),
988                    expression: expr.clone(),
989                    steps: vec![],
990                    eager: true,
991                    remove_on_failure: false,
992                }
993            ),
994            "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
995        );
996
997        // -- group L: DeclarativeDoTry -----------------------------------------
998        assert_eq!(
999            format!(
1000                "{:?}",
1001                BuilderStep::DeclarativeDoTry {
1002                    try_steps: vec![BuilderStep::Stop],
1003                    catch: vec![],
1004                    finally: None,
1005                }
1006            ),
1007            "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1008        );
1009
1010        // -- group M: DeclarativeStreamSplit -----------------------------------
1011        assert_eq!(
1012            format!(
1013                "{:?}",
1014                BuilderStep::DeclarativeStreamSplit {
1015                    stream_config: StreamSplitConfig {
1016                        format: StreamSplitFormat::Ndjson,
1017                        max_record_bytes: 1024 * 1024,
1018                        batch_size: 1,
1019                        chunk_size: None,
1020                        include_origin: true,
1021                    },
1022                    aggregation: AggregationStrategy::Original,
1023                    stop_on_exception: true,
1024                    steps: vec![BuilderStep::Stop],
1025                }
1026            ),
1027            "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] }"
1028        );
1029    }
1030
1031    #[test]
1032    fn test_builder_step_multicast_variant() {
1033        use camel_api::MulticastConfig;
1034
1035        let step = BuilderStep::Multicast {
1036            steps: vec![BuilderStep::To("direct:a".into())],
1037            config: MulticastConfig::new(),
1038        };
1039
1040        assert!(matches!(step, BuilderStep::Multicast { .. }));
1041    }
1042
1043    #[test]
1044    fn test_route_definition_defaults() {
1045        let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1046        assert_eq!(def.route_id(), "test-route");
1047        assert!(def.auto_startup());
1048        assert_eq!(def.startup_order(), 1000);
1049    }
1050
1051    #[test]
1052    fn test_route_definition_builders() {
1053        let def = RouteDefinition::new("direct:test", vec![])
1054            .with_route_id("my-route")
1055            .with_auto_startup(false)
1056            .with_startup_order(50);
1057        assert_eq!(def.route_id(), "my-route");
1058        assert!(!def.auto_startup());
1059        assert_eq!(def.startup_order(), 50);
1060    }
1061
1062    #[test]
1063    fn test_route_definition_accessors_cover_core_fields() {
1064        let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1065            .with_route_id("accessor-route");
1066
1067        assert_eq!(def.from_uri(), "direct:in");
1068        assert_eq!(def.steps().len(), 1);
1069        assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1070    }
1071
1072    #[test]
1073    fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1074        use camel_api::circuit_breaker::CircuitBreakerConfig;
1075        use camel_api::error_handler::ErrorHandlerConfig;
1076        use camel_component_api::ConcurrencyModel;
1077
1078        let def = RouteDefinition::new("direct:test", vec![])
1079            .with_route_id("eh-route")
1080            .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1081            .with_circuit_breaker(CircuitBreakerConfig::new())
1082            .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1083
1084        let eh = def
1085            .error_handler_config()
1086            .expect("error handler should be set");
1087        assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1088        assert!(def.circuit_breaker_config().is_some());
1089        assert!(matches!(
1090            def.concurrency_override(),
1091            Some(ConcurrencyModel::Concurrent { max: Some(4) })
1092        ));
1093    }
1094
1095    #[test]
1096    fn test_builder_step_debug_covers_many_variants() {
1097        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1098        use camel_api::{
1099            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1100            OpaqueProcessor, RoutingSlipConfig, Value,
1101        };
1102        use std::sync::Arc;
1103
1104        let expr = LanguageExpressionDef {
1105            language: "simple".into(),
1106            source: "${body}".into(),
1107        };
1108
1109        let steps = vec![
1110            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1111            BuilderStep::To("mock:out".into()),
1112            BuilderStep::Stop,
1113            BuilderStep::Log {
1114                level: camel_processor::LogLevel::Info,
1115                message: "hello".into(),
1116            },
1117            BuilderStep::DeclarativeSetHeader {
1118                key: "k".into(),
1119                value: ValueSourceDef::Literal(Value::String("v".into())),
1120            },
1121            BuilderStep::DeclarativeSetBody {
1122                value: ValueSourceDef::Expression(expr.clone()),
1123            },
1124            BuilderStep::DeclarativeFilter {
1125                predicate: expr.clone(),
1126                steps: vec![BuilderStep::Stop],
1127            },
1128            BuilderStep::DeclarativeChoice {
1129                whens: vec![DeclarativeWhenStep {
1130                    predicate: expr.clone(),
1131                    steps: vec![BuilderStep::Stop],
1132                }],
1133                otherwise: Some(vec![BuilderStep::Stop]),
1134            },
1135            BuilderStep::DeclarativeScript {
1136                expression: expr.clone(),
1137            },
1138            BuilderStep::DeclarativeSplit {
1139                expression: expr.clone(),
1140                aggregation: AggregationStrategy::Original,
1141                parallel: false,
1142                parallel_limit: Some(2),
1143                stop_on_exception: true,
1144                steps: vec![BuilderStep::Stop],
1145            },
1146            BuilderStep::Split {
1147                config: SplitterConfig::new(split_body_lines()),
1148                steps: vec![BuilderStep::Stop],
1149            },
1150            BuilderStep::Aggregate {
1151                config: camel_api::AggregatorConfig::correlate_by("id")
1152                    .complete_when_size(1)
1153                    .build()
1154                    .unwrap(),
1155            },
1156            BuilderStep::Filter {
1157                predicate: FilterPredicate::new(|_: &Exchange| true),
1158                steps: vec![BuilderStep::Stop],
1159            },
1160            BuilderStep::WireTap {
1161                uri: "mock:tap".into(),
1162            },
1163            BuilderStep::DeclarativeLog {
1164                level: camel_processor::LogLevel::Info,
1165                message: ValueSourceDef::Expression(expr.clone()),
1166            },
1167            BuilderStep::Bean {
1168                name: "bean".into(),
1169                method: "call".into(),
1170            },
1171            BuilderStep::Script {
1172                language: "rhai".into(),
1173                script: "body".into(),
1174            },
1175            BuilderStep::Throttle {
1176                config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1177                steps: vec![BuilderStep::Stop],
1178            },
1179            BuilderStep::LoadBalance {
1180                config: camel_api::LoadBalancerConfig::round_robin(),
1181                steps: vec![BuilderStep::To("mock:l1".into())],
1182            },
1183            BuilderStep::DynamicRouter {
1184                config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1185            },
1186            BuilderStep::RoutingSlip {
1187                config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1188            },
1189        ];
1190
1191        for step in steps {
1192            let dbg = format!("{step:?}");
1193            assert!(!dbg.is_empty());
1194        }
1195    }
1196
1197    #[test]
1198    fn test_route_definition_to_info_preserves_metadata() {
1199        let info = RouteDefinition::new("direct:test", vec![])
1200            .with_route_id("meta-route")
1201            .with_auto_startup(false)
1202            .with_startup_order(7)
1203            .to_info();
1204
1205        assert_eq!(info.route_id(), "meta-route");
1206        assert!(!info.auto_startup());
1207        assert_eq!(info.startup_order(), 7);
1208    }
1209
1210    #[test]
1211    fn test_choice_builder_step_debug() {
1212        use camel_api::FilterPredicate;
1213
1214        fn always_true(_: &camel_api::Exchange) -> bool {
1215            true
1216        }
1217
1218        let step = BuilderStep::Choice {
1219            whens: vec![WhenStep {
1220                predicate: FilterPredicate::new(always_true),
1221                steps: vec![BuilderStep::To("mock:a".into())],
1222            }],
1223            otherwise: None,
1224        };
1225        let debug = format!("{step:?}");
1226        assert!(debug.contains("Choice"));
1227    }
1228
1229    #[test]
1230    fn test_route_definition_unit_of_work() {
1231        use camel_api::UnitOfWorkConfig;
1232        let config = UnitOfWorkConfig {
1233            on_complete: Some("log:complete".into()),
1234            on_failure: Some("log:failed".into()),
1235        };
1236        let def = RouteDefinition::new("direct:test", vec![])
1237            .with_route_id("uow-test")
1238            .with_unit_of_work(config.clone());
1239        assert_eq!(
1240            def.unit_of_work_config().unwrap().on_complete.as_deref(),
1241            Some("log:complete")
1242        );
1243        assert_eq!(
1244            def.unit_of_work_config().unwrap().on_failure.as_deref(),
1245            Some("log:failed")
1246        );
1247
1248        let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1249        assert!(def_no_uow.unit_of_work_config().is_none());
1250    }
1251
1252    #[test]
1253    fn test_route_definition_security_policy_accessor() {
1254        use async_trait::async_trait;
1255        use camel_api::CamelError;
1256        use camel_api::Exchange;
1257        use camel_api::security_policy::{
1258            AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1259        };
1260
1261        struct StubPolicy;
1262        #[async_trait]
1263        impl SecurityPolicy for StubPolicy {
1264            async fn evaluate(
1265                &self,
1266                _exchange: &mut Exchange,
1267            ) -> Result<AuthorizationDecision, CamelError> {
1268                Ok(AuthorizationDecision::Granted {
1269                    principal: Principal {
1270                        subject: "test".into(),
1271                        issuer: "test".into(),
1272                        audience: vec![],
1273                        scopes: vec![],
1274                        roles: vec![],
1275                        claims: serde_json::Value::Null,
1276                    },
1277                })
1278            }
1279        }
1280
1281        let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1282        assert!(def_no_sp.security_policy_config().is_none());
1283
1284        let def = RouteDefinition::new("direct:test", vec![])
1285            .with_route_id("sp-test")
1286            .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1287        assert!(def.security_policy_config().is_some());
1288    }
1289
1290    #[test]
1291    fn test_route_definition_security_authenticator_accessor() {
1292        use camel_api::security_policy::Principal;
1293
1294        struct TestAuth;
1295        #[async_trait::async_trait]
1296        impl TokenAuthenticator for TestAuth {
1297            async fn authenticate_bearer(
1298                &self,
1299                _token: &str,
1300            ) -> Result<Principal, camel_api::CamelError> {
1301                Ok(Principal {
1302                    subject: "test".into(),
1303                    issuer: "test".into(),
1304                    audience: vec![],
1305                    scopes: vec![],
1306                    roles: vec![],
1307                    claims: serde_json::Value::Null,
1308                })
1309            }
1310        }
1311
1312        let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1313        assert!(def_no_auth.security_authenticator().is_none());
1314
1315        let auth = Arc::new(TestAuth);
1316        let def = RouteDefinition::new("direct:test".to_string(), vec![])
1317            .with_security_authenticator(auth);
1318        assert!(def.security_authenticator().is_some());
1319    }
1320}
camel_core/lifecycle/application/route_definition.rs

camel_core/lifecycle/application/
route_definition.rs
