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

camel_core/lifecycle/application/
route_definition.rs
