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    SpanKindHint, 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    /// Declarative remove_header, resolved at route-add time.
85    DeclarativeRemoveHeader {
86        key: String,
87    },
88    DeclarativeSetProperty {
89        key: String,
90        value_source: ValueSourceDef,
91    },
92    /// Declarative set_body (literal or language-based value), resolved at route-add time.
93    DeclarativeSetBody {
94        value: ValueSourceDef,
95    },
96    /// Declarative filter using a language predicate, resolved at route-add time.
97    DeclarativeFilter {
98        predicate: LanguageExpressionDef,
99        steps: Vec<BuilderStep>,
100    },
101    /// Declarative choice/when/otherwise using language predicates, resolved at route-add time.
102    DeclarativeChoice {
103        whens: Vec<DeclarativeWhenStep>,
104        otherwise: Option<Vec<BuilderStep>>,
105    },
106    /// Declarative script step evaluated by language and written to body.
107    DeclarativeScript {
108        expression: LanguageExpressionDef,
109    },
110    DeclarativeFunction {
111        definition: camel_api::FunctionDefinition,
112    },
113    /// Declarative split using a language expression, resolved at route-add time.
114    DeclarativeSplit {
115        expression: LanguageExpressionDef,
116        aggregation: camel_api::splitter::AggregationStrategy,
117        parallel: bool,
118        parallel_limit: Option<usize>,
119        stop_on_exception: bool,
120        steps: Vec<BuilderStep>,
121    },
122    /// Declarative stream split using a streaming split expression, resolved at route-add time.
123    DeclarativeStreamSplit {
124        stream_config: camel_api::StreamSplitConfig,
125        aggregation: camel_api::splitter::AggregationStrategy,
126        stop_on_exception: bool,
127        steps: Vec<BuilderStep>,
128    },
129    DeclarativeDynamicRouter {
130        expression: LanguageExpressionDef,
131        uri_delimiter: String,
132        cache_size: i32,
133        ignore_invalid_endpoints: bool,
134        max_iterations: usize,
135    },
136    DeclarativeRoutingSlip {
137        expression: LanguageExpressionDef,
138        uri_delimiter: String,
139        cache_size: i32,
140        ignore_invalid_endpoints: bool,
141    },
142    /// A Splitter sub-pipeline: config + nested steps to execute per fragment.
143    Split {
144        config: SplitterConfig,
145        steps: Vec<BuilderStep>,
146    },
147    /// An Aggregator step: collects exchanges by correlation key, emits when complete.
148    Aggregate {
149        config: AggregatorConfig,
150    },
151    /// A Filter sub-pipeline: predicate + nested steps executed only when predicate is true.
152    Filter {
153        predicate: FilterPredicate,
154        steps: Vec<BuilderStep>,
155    },
156    /// A Choice step: evaluates when-clauses in order, routes to the first match.
157    /// If no when matches, the optional otherwise branch is used.
158    Choice {
159        whens: Vec<WhenStep>,
160        otherwise: Option<Vec<BuilderStep>>,
161    },
162    /// A WireTap step: sends a clone of the exchange to a tap endpoint (fire-and-forget).
163    WireTap {
164        uri: String,
165    },
166    /// A Multicast step: sends the same exchange to multiple destinations.
167    Multicast {
168        steps: Vec<BuilderStep>,
169        config: MulticastConfig,
170    },
171    /// Declarative log step with a language-evaluated message, resolved at route-add time.
172    DeclarativeLog {
173        level: camel_processor::LogLevel,
174        message: ValueSourceDef,
175    },
176    /// Bean invocation step — resolved at route-add time.
177    Bean {
178        name: String,
179        method: String,
180    },
181    /// Script step: executes a script that can mutate the exchange.
182    /// The script has access to `headers`, `properties`, and `body`.
183    Script {
184        language: String,
185        script: String,
186    },
187    /// Throttle step: rate limiting with configurable behavior when limit exceeded.
188    Throttle {
189        config: camel_api::ThrottlerConfig,
190        steps: Vec<BuilderStep>,
191    },
192    /// LoadBalance step: distributes exchanges across multiple endpoints using a strategy.
193    LoadBalance {
194        config: camel_api::LoadBalancerConfig,
195        steps: Vec<BuilderStep>,
196    },
197    /// DynamicRouter step: routes exchanges dynamically based on expression evaluation.
198    DynamicRouter {
199        config: camel_api::DynamicRouterConfig,
200    },
201    RoutingSlip {
202        config: camel_api::RoutingSlipConfig,
203    },
204    RecipientList {
205        config: camel_api::recipient_list::RecipientListConfig,
206    },
207    DeclarativeRecipientList {
208        expression: LanguageExpressionDef,
209        delimiter: String,
210        parallel: bool,
211        parallel_limit: Option<usize>,
212        stop_on_exception: bool,
213        aggregation: String,
214    },
215    Delay {
216        config: camel_api::DelayConfig,
217    },
218    /// Runtime loop with closure-based predicate (programmatic DSL).
219    Loop {
220        config: LoopConfig,
221        steps: Vec<BuilderStep>,
222    },
223    /// Declarative loop with optional language-based while predicate (YAML DSL).
224    DeclarativeLoop {
225        count: Option<usize>,
226        while_predicate: Option<LanguageExpressionDef>,
227        steps: Vec<BuilderStep>,
228        max_iterations: Option<usize>,
229    },
230    /// EIP-7 enrich: synchronous content enrichment via a resolved producer.
231    Enrich {
232        uri: String,
233        strategy: Option<String>,
234        timeout_ms: Option<u64>,
235    },
236    /// EIP-7 pollEnrich: blocking poll of a PollingConsumer with timeout.
237    PollEnrich {
238        uri: String,
239        strategy: Option<String>,
240        timeout_ms: Option<u64>,
241    },
242    /// Validate step: evaluates a language expression as predicate.
243    /// Exchange passes if predicate returns true; else CamelError::ValidationError.
244    Validate {
245        predicate: LanguageExpressionDef,
246    },
247    /// Claim Check step (EIP). Transforms the exchange body to/from a
248    /// `ClaimCheckRepository` by key. Process-mode, no child pipeline.
249    /// `filter` enables selective merge-back of body/headers during checkout.
250    ClaimCheck {
251        repository: String,
252        operation: String,
253        key: LanguageExpressionDef,
254        filter: Option<String>,
255    },
256    /// Sampling step (EIP). Passes 1 of every N exchanges (counter-based,
257    /// deterministic). Non-sampled exchanges get CamelStop=true (drop semantics).
258    /// Process-mode, stateless. No StepLifecycle — counter is route-scoped.
259    Sampling {
260        period: usize,
261    },
262    /// Sort step (EIP). Orders a body array by extracting a sort key
263    /// from each element via a language expression. Process-mode, stateless.
264    Sort {
265        expression: LanguageExpressionDef,
266        reverse: bool,
267    },
268    /// Idempotent Consumer step (EIP). Wraps a child sub-pipeline that runs
269    /// only when the message-id is NOT present in the named repository.
270    /// Compiled to a `IdempotentConsumerSegment` (OutcomePipeline, segment-mode).
271    IdempotentConsumer {
272        repository: String,
273        expression: LanguageExpressionDef,
274        steps: Vec<BuilderStep>,
275        eager: bool,
276        remove_on_failure: bool,
277    },
278    /// Cache step (EIP). Wraps a child sub-pipeline that runs only on cache
279    /// miss; on hit the stored body is reconstructed and the child is skipped.
280    /// Compiled to a `CacheService` (OutcomePipeline, segment-mode).
281    Cache {
282        repository: Option<String>,
283        key: LanguageExpressionDef,
284        ttl: Option<String>,
285        max_entry_bytes: Option<usize>,
286        /// Coalesce concurrent misses on the same key into one `on_miss` run.
287        coalesce_misses: bool,
288        on_miss: Vec<BuilderStep>,
289    },
290    /// Cache Invalidate step (EIP). Removes a single cache entry by exact key
291    /// (`key`) or every entry under a namespace prefix (`key_prefix`) — exactly
292    /// one of the two. Compiled to a `CacheInvalidateService` (OutcomePipeline,
293    /// segment-mode).
294    CacheInvalidate {
295        repository: Option<String>,
296        key: Option<LanguageExpressionDef>,
297        key_prefix: Option<LanguageExpressionDef>,
298    },
299    /// Cache Clear step (EIP). Removes all entries from the repository.
300    /// Compiled to a `CacheClearService` (OutcomePipeline, segment-mode).
301    CacheClear {
302        repository: Option<String>,
303    },
304    /// Cache Stats step (EIP). Emits the repository stats as a JSON body.
305    /// Compiled to a `CacheStatsService` (OutcomePipeline, segment-mode).
306    CacheStats {
307        repository: Option<String>,
308    },
309    /// Cache Peek Stale step (EIP). Serves a stale (post-expiry) cache entry.
310    /// Compiled to a `CachePeekStaleService` (OutcomePipeline, segment-mode).
311    CachePeekStale {
312        repository: Option<String>,
313        key: LanguageExpressionDef,
314        on_miss: camel_processor::PeekStaleMissPolicy,
315    },
316    /// Declarative doTry/doCatch/doFinally, resolved at route-add time.
317    DeclarativeDoTry {
318        try_steps: Vec<BuilderStep>,
319        catch: Vec<DoTryCatchClauseBuilder>,
320        finally: Option<DoTryFinallyBuilder>,
321    },
322    /// Resequencer EIP: resequences exchanges by sequence number.
323    /// Must be a top-level step (not nested inside structural EIPs).
324    Resequence {
325        policy_config: ResequencePolicyConfig,
326    },
327}
328
329impl BuilderStep {
330    /// Stable span name for this step (span-name-enrichment, task 1.2).
331    ///
332    /// `To` yields `to:{scheme}` where scheme is the AUTHORED URI before the
333    /// first `:` — a pure string split, no endpoint resolution, so a `SkipTo`
334    /// interception never rewrites the label. EIP variants yield their
335    /// kebab-case EIP name; declarative twins share the label of their
336    /// programmatic sibling (`DeclarativeSplit` and `Split` are both
337    /// `"split"`). Genuinely anonymous variants (opaque processors, the Stop
338    /// marker) yield `None` — spans fall back to the positional
339    /// `step-{index}` name.
340    ///
341    /// The match is exhaustive by design (no catch-all) so a future variant
342    /// forces a label decision here.
343    pub(crate) fn span_label(&self) -> Option<String> {
344        match self {
345            // Anonymous.
346            Self::Processor(_) | Self::Stop => None,
347
348            Self::To(uri) => uri.contains(':').then(|| {
349                let scheme = uri.split(':').next().unwrap_or_default();
350                format!("to:{scheme}")
351            }),
352
353            Self::Log { .. } | Self::DeclarativeLog { .. } => Some("log".into()),
354
355            Self::DeclarativeSetHeader { .. } => Some("set-header".into()),
356            Self::DeclarativeSetHeaderIfAbsent { .. } => Some("set-header-if-absent".into()),
357            Self::DeclarativeRemoveHeader { .. } => Some("remove-header".into()),
358            Self::DeclarativeSetProperty { .. } => Some("set-property".into()),
359            Self::DeclarativeSetBody { .. } => Some("set-body".into()),
360
361            Self::DeclarativeFilter { .. } | Self::Filter { .. } => Some("filter".into()),
362            Self::DeclarativeChoice { .. } | Self::Choice { .. } => Some("choice".into()),
363            Self::DeclarativeScript { .. } | Self::Script { .. } => Some("script".into()),
364            Self::DeclarativeFunction { .. } => Some("function".into()),
365
366            Self::DeclarativeSplit { .. }
367            | Self::DeclarativeStreamSplit { .. }
368            | Self::Split { .. } => Some("split".into()),
369
370            Self::DeclarativeDynamicRouter { .. } | Self::DynamicRouter { .. } => {
371                Some("dynamic-router".into())
372            }
373            Self::DeclarativeRoutingSlip { .. } | Self::RoutingSlip { .. } => {
374                Some("routing-slip".into())
375            }
376            Self::DeclarativeRecipientList { .. } | Self::RecipientList { .. } => {
377                Some("recipient-list".into())
378            }
379
380            Self::Aggregate { .. } => Some("aggregate".into()),
381            Self::WireTap { .. } => Some("wire-tap".into()),
382            Self::Multicast { .. } => Some("multicast".into()),
383            Self::Bean { .. } => Some("bean".into()),
384            Self::Throttle { .. } => Some("throttle".into()),
385            Self::LoadBalance { .. } => Some("load-balance".into()),
386            Self::Delay { .. } => Some("delay".into()),
387            Self::Loop { .. } | Self::DeclarativeLoop { .. } => Some("loop".into()),
388            Self::Enrich { .. } => Some("enrich".into()),
389            Self::PollEnrich { .. } => Some("poll-enrich".into()),
390            Self::Validate { .. } => Some("validate".into()),
391            Self::ClaimCheck { .. } => Some("claim-check".into()),
392            Self::Sampling { .. } => Some("sampling".into()),
393            Self::Sort { .. } => Some("sort".into()),
394            Self::IdempotentConsumer { .. } => Some("idempotent-consumer".into()),
395            Self::Cache { .. } => Some("cache".into()),
396            Self::CacheInvalidate { .. } => Some("cache-invalidate".into()),
397            Self::CacheClear { .. } => Some("cache-clear".into()),
398            Self::CacheStats { .. } => Some("cache-stats".into()),
399            Self::CachePeekStale { .. } => Some("cache-peek-stale".into()),
400            Self::DeclarativeDoTry { .. } => Some("do-try".into()),
401            Self::Resequence { .. } => Some("resequence".into()),
402        }
403    }
404
405    /// Span kind hint for this step's step span (span-kind-hint, task 1.2).
406    ///
407    /// Endpoint-bearing variants (`To`, `Enrich`, `PollEnrich`, `WireTap`)
408    /// classify the AUTHORED URI scheme — the text before the first `:`,
409    /// compared with `eq_ignore_ascii_case`, no endpoint resolution — so a
410    /// `SkipTo` interception never rewrites the kind. Messaging broker
411    /// schemes map to `Producer` (async, one-way send) and synchronous
412    /// outbound protocols to `Client`; every other scheme and scheme-less
413    /// URIs stay `Internal` route processing.
414    ///
415    /// Every non-endpoint variant is internal route processing. The match is
416    /// exhaustive by design (no `_` wildcard): adding a `BuilderStep` variant
417    /// without an explicit span-kind decision here is a compile error
418    /// (rc-4mz7). `BuilderStep` is defined in this crate and carries no
419    /// `#[non_exhaustive]`, so same-crate exhaustiveness holds.
420    pub(crate) fn span_kind_hint(&self) -> SpanKindHint {
421        match self {
422            // Endpoint-bearing variants: classify by the authored scheme.
423            Self::To(uri)
424            | Self::Enrich { uri, .. }
425            | Self::PollEnrich { uri, .. }
426            | Self::WireTap { uri, .. } => uri_span_kind(uri),
427
428            // rc-4mz7: every remaining variant is named explicitly — no
429            // wildcard. A new BuilderStep variant fails to compile until it
430            // is classified here.
431            Self::Processor(..)
432            | Self::Stop
433            | Self::Log { .. }
434            | Self::DeclarativeSetHeader { .. }
435            | Self::DeclarativeSetHeaderIfAbsent { .. }
436            | Self::DeclarativeRemoveHeader { .. }
437            | Self::DeclarativeSetProperty { .. }
438            | Self::DeclarativeSetBody { .. }
439            | Self::DeclarativeFilter { .. }
440            | Self::DeclarativeChoice { .. }
441            | Self::DeclarativeScript { .. }
442            | Self::DeclarativeFunction { .. }
443            | Self::DeclarativeSplit { .. }
444            | Self::DeclarativeStreamSplit { .. }
445            | Self::DeclarativeDynamicRouter { .. }
446            | Self::DeclarativeRoutingSlip { .. }
447            | Self::Split { .. }
448            | Self::Aggregate { .. }
449            | Self::Filter { .. }
450            | Self::Choice { .. }
451            | Self::Multicast { .. }
452            | Self::DeclarativeLog { .. }
453            | Self::Bean { .. }
454            | Self::Script { .. }
455            | Self::Throttle { .. }
456            | Self::LoadBalance { .. }
457            | Self::DynamicRouter { .. }
458            | Self::RoutingSlip { .. }
459            | Self::RecipientList { .. }
460            | Self::DeclarativeRecipientList { .. }
461            | Self::Delay { .. }
462            | Self::Loop { .. }
463            | Self::DeclarativeLoop { .. }
464            | Self::Validate { .. }
465            | Self::ClaimCheck { .. }
466            | Self::Sampling { .. }
467            | Self::Sort { .. }
468            | Self::IdempotentConsumer { .. }
469            | Self::Cache { .. }
470            | Self::CacheInvalidate { .. }
471            | Self::CacheClear { .. }
472            | Self::CacheStats { .. }
473            | Self::CachePeekStale { .. }
474            | Self::DeclarativeDoTry { .. }
475            | Self::Resequence { .. } => SpanKindHint::Internal,
476        }
477    }
478}
479
480/// Classify an AUTHORED endpoint URI by its scheme — the text before the
481/// first `:`, compared with `eq_ignore_ascii_case`, no endpoint resolution.
482///
483/// Messaging broker schemes map to `Producer` (async, one-way send) and
484/// synchronous outbound protocols to `Client`; every other scheme and
485/// scheme-less URIs stay `Internal` — the kind is never guessed.
486fn uri_span_kind(uri: &str) -> SpanKindHint {
487    /// Broker-style destinations: an async, one-way send.
488    const PRODUCER_SCHEMES: [&str; 5] = ["kafka", "jms", "activemq", "artemis", "mqtt"];
489    /// Synchronous outbound request/response protocols.
490    const CLIENT_SCHEMES: [&str; 12] = [
491        "http",
492        "https",
493        "grpc",
494        "grpcs",
495        "ws",
496        "redis",
497        "opensearch",
498        "sql",
499        "surrealdb",
500        "cxf",
501        "llm",
502        "mcp",
503    ];
504
505    let scheme = uri.split(':').next().unwrap_or_default();
506    if PRODUCER_SCHEMES
507        .iter()
508        .any(|s| s.eq_ignore_ascii_case(scheme))
509    {
510        SpanKindHint::Producer
511    } else if CLIENT_SCHEMES
512        .iter()
513        .any(|s| s.eq_ignore_ascii_case(scheme))
514    {
515        SpanKindHint::Client
516    } else {
517        SpanKindHint::Internal
518    }
519}
520
521/// An unresolved route definition. "to" URIs have not been resolved to producers yet.
522pub struct RouteDefinition {
523    pub(crate) from_uri: String,
524    pub(crate) steps: Vec<BuilderStep>,
525    /// Optional per-route error handler config. Takes precedence over the global one.
526    pub(crate) error_handler: Option<ErrorHandlerConfig>,
527    /// Optional circuit breaker config. Applied between error handler and step pipeline.
528    pub(crate) circuit_breaker: Option<CircuitBreakerConfig>,
529    /// Circuit breaker fallback sub-pipeline, as UNRESOLVED steps.
530    ///
531    /// Sibling of [`RouteDefinition::circuit_breaker`]: the DSL layers thread
532    /// `Vec<BuilderStep>` here (same rule as `cache_peek_stale.on_miss`) and
533    /// camel-core compiles it via the `StepCompilerRegistry` when the route is
534    /// compiled. The resolved `CircuitBreakerConfig.fallback` (`BoxProcessor`)
535    /// stays `None` until that compile — the DSL never constructs processors.
536    pub(crate) circuit_breaker_fallback: Vec<BuilderStep>,
537    pub(crate) security_policy: Option<SecurityPolicyConfig>,
538    /// Optional token authenticator for validating JWT/OAuth tokens.
539    pub(crate) security_authenticator: Option<Arc<dyn TokenAuthenticator>>,
540    /// Named authentication providers for this route, injected into the
541    /// consumer `SecurityContext` so Phase-2 transports can resolve them.
542    pub(crate) provider_registry: Option<Arc<camel_auth::ProviderRegistry>>,
543    /// Declared provider name for plan compilation (`security_policy.provider`
544    /// in the DSL). Resolved against the provider registry at staging time;
545    /// a missing name fails compilation instead of downgrading to `Public`.
546    pub(crate) security_provider: Option<String>,
547    /// Route-level audience override for plan compilation. When present the
548    /// compiled `RouteSecurityPlan.audience_binding` uses these audiences
549    /// (issuers still come from the provider); when absent the resolved
550    /// provider's binding is copied verbatim.
551    pub(crate) security_audiences: Option<Vec<String>>,
552    /// Optional Unit of Work config for in-flight tracking and completion hooks.
553    pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
554    /// User override for the consumer's concurrency model. `None` means
555    /// "use whatever the consumer declares".
556    pub(crate) concurrency: Option<ConcurrencyModel>,
557    /// Unique identifier for this route. Required.
558    pub(crate) route_id: String,
559    /// Whether this route should start automatically when the context starts.
560    pub(crate) auto_startup: bool,
561    /// Order in which routes are started. Lower values start first.
562    pub(crate) startup_order: i32,
563    pub(crate) source_hash: Option<u64>,
564}
565
566impl RouteDefinition {
567    /// Create a new route definition with the required route ID.
568    pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
569        Self {
570            from_uri: from_uri.into(),
571            steps,
572            error_handler: None,
573            circuit_breaker: None,
574            circuit_breaker_fallback: Vec::new(),
575            security_policy: None,
576            security_authenticator: None,
577            provider_registry: None,
578            security_provider: None,
579            security_audiences: None,
580            unit_of_work: None,
581            concurrency: None,
582            route_id: String::new(), // Will be set by with_route_id()
583            auto_startup: true,
584            startup_order: 1000,
585            source_hash: None,
586        }
587    }
588
589    /// The source endpoint URI.
590    pub fn from_uri(&self) -> &str {
591        &self.from_uri
592    }
593
594    /// The steps in this route definition.
595    pub fn steps(&self) -> &[BuilderStep] {
596        &self.steps
597    }
598
599    pub fn circuit_breaker_fallback(&self) -> &[BuilderStep] {
600        &self.circuit_breaker_fallback
601    }
602
603    /// Transform the step list, consuming and returning self.
604    /// Used for post-parse instrumentation (e.g. benchmark timing injection
605    /// around `To` steps). The field stays private; callers rebuild via this
606    /// consuming method rather than holding a `&mut` alias.
607    pub fn map_steps(mut self, f: impl FnOnce(Vec<BuilderStep>) -> Vec<BuilderStep>) -> Self {
608        self.steps = f(self.steps);
609        self
610    }
611
612    /// Set a per-route error handler, overriding the global one.
613    pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
614        self.error_handler = Some(config);
615        self
616    }
617
618    /// Get the route-level error handler config, if set.
619    pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
620        self.error_handler.as_ref()
621    }
622
623    /// Set a circuit breaker for this route.
624    pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
625        self.circuit_breaker = Some(config);
626        self
627    }
628
629    /// Set the circuit breaker fallback sub-pipeline (unresolved steps).
630    ///
631    /// Mirror of [`RouteDefinition::with_circuit_breaker`]: the steps are
632    /// compiled by camel-core at route-compile time (registry monopoly) and
633    /// attached to the resolved `CircuitBreakerConfig.fallback`. Empty when no
634    /// CB is configured or the CB has no fallback.
635    pub fn with_circuit_breaker_fallback(mut self, steps: Vec<BuilderStep>) -> Self {
636        self.circuit_breaker_fallback = steps;
637        self
638    }
639
640    /// Set a security policy for this route.
641    pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
642        self.security_policy = Some(config);
643        self
644    }
645
646    /// Set a token authenticator for this route.
647    pub fn with_security_authenticator(
648        mut self,
649        authenticator: Arc<dyn TokenAuthenticator>,
650    ) -> Self {
651        self.security_authenticator = Some(authenticator);
652        self
653    }
654
655    /// Set the named authentication providers for this route.
656    ///
657    /// The registry is injected into the consumer `SecurityContext` so
658    /// Phase-2 transports can resolve providers without holding their own
659    /// authenticator.
660    pub fn with_provider_registry(mut self, registry: Arc<camel_auth::ProviderRegistry>) -> Self {
661        self.provider_registry = Some(registry);
662        self
663    }
664
665    /// Set the declared provider name used by plan compilation.
666    pub fn with_security_provider(mut self, name: impl Into<String>) -> Self {
667        self.security_provider = Some(name.into());
668        self
669    }
670
671    /// Set the route-level audience override used by plan compilation.
672    pub fn with_security_audiences(mut self, audiences: Vec<String>) -> Self {
673        self.security_audiences = Some(audiences);
674        self
675    }
676
677    /// The declared provider name, if any.
678    pub fn security_provider(&self) -> Option<&str> {
679        self.security_provider.as_deref()
680    }
681
682    /// The route-level audience override, if any.
683    pub fn security_audiences(&self) -> Option<&[String]> {
684        self.security_audiences.as_deref()
685    }
686
687    /// Set a unit of work config for this route.
688    pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
689        self.unit_of_work = Some(config);
690        self
691    }
692
693    /// Get the unit of work config, if set.
694    pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
695        self.unit_of_work.as_ref()
696    }
697
698    /// Get the circuit breaker config, if set.
699    pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
700        self.circuit_breaker.as_ref()
701    }
702
703    pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
704        self.security_policy.as_ref()
705    }
706
707    pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
708        self.security_authenticator.as_ref()
709    }
710
711    /// User-specified concurrency override, if any.
712    pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
713        self.concurrency.as_ref()
714    }
715
716    /// Override the consumer's concurrency model for this route.
717    pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
718        self.concurrency = Some(model);
719        self
720    }
721
722    /// Get the route ID.
723    pub fn route_id(&self) -> &str {
724        &self.route_id
725    }
726
727    /// Whether this route should start automatically when the context starts.
728    pub fn auto_startup(&self) -> bool {
729        self.auto_startup
730    }
731
732    /// Order in which routes are started. Lower values start first.
733    pub fn startup_order(&self) -> i32 {
734        self.startup_order
735    }
736
737    /// Set a unique identifier for this route.
738    pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
739        self.route_id = id.into();
740        self
741    }
742
743    /// Set whether this route should start automatically.
744    pub fn with_auto_startup(mut self, auto: bool) -> Self {
745        self.auto_startup = auto;
746        self
747    }
748
749    /// Set the startup order. Lower values start first.
750    pub fn with_startup_order(mut self, order: i32) -> Self {
751        self.startup_order = order;
752        self
753    }
754
755    pub fn with_source_hash(mut self, hash: u64) -> Self {
756        self.source_hash = Some(hash);
757        self
758    }
759
760    pub fn source_hash(&self) -> Option<u64> {
761        self.source_hash
762    }
763
764    /// Extract the metadata fields needed for introspection.
765    /// This is used by RouteController to store route info without the non-Sync steps.
766    pub fn to_info(&self) -> RouteDefinitionInfo {
767        RouteDefinitionInfo {
768            route_id: self.route_id.clone(),
769            auto_startup: self.auto_startup,
770            startup_order: self.startup_order,
771            source_hash: self.source_hash,
772        }
773    }
774}
775
776/// Minimal route definition metadata for introspection.
777///
778/// This struct contains only the metadata fields from [`RouteDefinition`]
779/// that are needed for route lifecycle management, without the `steps` field
780/// (which contains non-Sync types and cannot be stored in a Sync struct).
781#[derive(Clone)]
782pub struct RouteDefinitionInfo {
783    route_id: String,
784    auto_startup: bool,
785    startup_order: i32,
786    pub(crate) source_hash: Option<u64>,
787}
788
789impl RouteDefinitionInfo {
790    /// Get the route ID.
791    pub fn route_id(&self) -> &str {
792        &self.route_id
793    }
794
795    /// Whether this route should start automatically when the context starts.
796    pub fn auto_startup(&self) -> bool {
797        self.auto_startup
798    }
799
800    /// Order in which routes are started. Lower values start first.
801    pub fn startup_order(&self) -> i32 {
802        self.startup_order
803    }
804
805    pub fn source_hash(&self) -> Option<u64> {
806        self.source_hash
807    }
808}
809
810#[cfg(test)]
811mod tests {
812    use super::*;
813
814    /// Task 1.2 (span-name-enrichment): `BuilderStep::span_label` mapping.
815    ///
816    /// `To` labels carry the authored URI scheme (before the first `:`, no
817    /// interception resolution); EIP variants map to kebab-case EIP names
818    /// (declarative twins share the label of their programmatic sibling);
819    /// anonymous variants (opaque processors) carry no label.
820    #[test]
821    fn builder_step_span_label_mapping() {
822        use camel_api::declarative::LanguageExpressionDef;
823        use camel_api::splitter::AggregationStrategy;
824        use camel_api::{BoxProcessor, IdentityProcessor, OpaqueProcessor};
825
826        let expr = LanguageExpressionDef {
827            language: "simple".into(),
828            source: "${body}".into(),
829        };
830
831        // To: scheme is the authored URI before the first ':'.
832        assert_eq!(
833            BuilderStep::To("direct:tree-sub".into())
834                .span_label()
835                .as_deref(),
836            Some("to:direct")
837        );
838        assert_eq!(
839            BuilderStep::To("http://api.example/x".into())
840                .span_label()
841                .as_deref(),
842            Some("to:http")
843        );
844        // Scheme-less URI: no label — never leak the full text.
845        assert_eq!(BuilderStep::To("garbage".into()).span_label(), None);
846
847        // EIP variants → kebab-case EIP name.
848        assert_eq!(
849            BuilderStep::Log {
850                level: camel_processor::LogLevel::Info,
851                message: "m".into(),
852            }
853            .span_label()
854            .as_deref(),
855            Some("log")
856        );
857        assert_eq!(
858            BuilderStep::Split {
859                config: camel_api::splitter::SplitterConfig::new(
860                    camel_api::splitter::split_body_lines()
861                ),
862                steps: vec![BuilderStep::Stop],
863            }
864            .span_label()
865            .as_deref(),
866            Some("split")
867        );
868        assert_eq!(
869            BuilderStep::DeclarativeSplit {
870                expression: expr.clone(),
871                aggregation: AggregationStrategy::Original,
872                parallel: false,
873                parallel_limit: None,
874                stop_on_exception: true,
875                steps: vec![BuilderStep::Stop],
876            }
877            .span_label()
878            .as_deref(),
879            Some("split")
880        );
881        assert_eq!(
882            BuilderStep::DeclarativeStreamSplit {
883                stream_config: camel_api::StreamSplitConfig::default(),
884                aggregation: AggregationStrategy::Original,
885                stop_on_exception: true,
886                steps: vec![BuilderStep::Stop],
887            }
888            .span_label()
889            .as_deref(),
890            Some("split")
891        );
892        assert_eq!(BuilderStep::Stop.span_label(), None);
893
894        // Anonymous variants → None.
895        assert_eq!(
896            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
897                .span_label(),
898            None
899        );
900    }
901
902    /// Task 1.2 (span-kind-hint): `BuilderStep::span_kind_hint` mapping.
903    ///
904    /// Endpoint-bearing variants (`To`, `Enrich`, `PollEnrich`, `WireTap`)
905    /// classify the AUTHORED URI scheme (text before the first `:`,
906    /// compared case-insensitively, no endpoint resolution): messaging
907    /// brokers map to `Producer`, synchronous outbound protocols to
908    /// `Client`. Every other scheme, scheme-less URIs, and every
909    /// non-endpoint variant stay `Internal` route processing. The
910    /// exhaustive match (no wildcard) in `span_kind_hint` is the rc-4mz7
911    /// contract: a new `BuilderStep` variant cannot compile without a
912    /// classification here.
913    #[test]
914    fn builder_step_span_kind_hint_mapping() {
915        use camel_api::splitter::split_body_lines;
916        use camel_api::{Exchange, FilterPredicate, SpanKindHint};
917
918        let kind = |uri: &str| BuilderStep::To(uri.into()).span_kind_hint();
919
920        // Messaging brokers → Producer.
921        assert_eq!(kind("kafka:orders"), SpanKindHint::Producer);
922        assert_eq!(kind("jms:q"), SpanKindHint::Producer);
923        assert_eq!(kind("activemq:q"), SpanKindHint::Producer);
924        assert_eq!(kind("artemis:q"), SpanKindHint::Producer);
925        assert_eq!(kind("mqtt:t"), SpanKindHint::Producer);
926        // Scheme comparison is case-insensitive.
927        assert_eq!(kind("KAFKA:orders"), SpanKindHint::Producer);
928
929        // Synchronous outbound protocols → Client.
930        assert_eq!(kind("http://x"), SpanKindHint::Client);
931        assert_eq!(kind("https://x"), SpanKindHint::Client);
932        assert_eq!(kind("grpc://x"), SpanKindHint::Client);
933        assert_eq!(kind("grpcs://x"), SpanKindHint::Client);
934        assert_eq!(kind("ws://x"), SpanKindHint::Client);
935        assert_eq!(kind("redis://x"), SpanKindHint::Client);
936        assert_eq!(kind("opensearch://x"), SpanKindHint::Client);
937        assert_eq!(kind("sql:db"), SpanKindHint::Client);
938        assert_eq!(kind("surrealdb://x"), SpanKindHint::Client);
939        assert_eq!(kind("cxf://x"), SpanKindHint::Client);
940        assert_eq!(kind("llm://x"), SpanKindHint::Client);
941        assert_eq!(kind("mcp://x"), SpanKindHint::Client);
942
943        // In-process and unknown schemes → Internal.
944        assert_eq!(kind("direct:y"), SpanKindHint::Internal);
945        assert_eq!(kind("timer:z"), SpanKindHint::Internal);
946        // Scheme-less URI: the kind is never guessed.
947        assert_eq!(kind("garbage"), SpanKindHint::Internal);
948
949        // Non-`To` variants are internal route processing.
950        assert_eq!(
951            BuilderStep::Log {
952                level: camel_processor::LogLevel::Info,
953                message: "m".into(),
954            }
955            .span_kind_hint(),
956            SpanKindHint::Internal
957        );
958        assert_eq!(
959            BuilderStep::Filter {
960                predicate: FilterPredicate::new(|_: &Exchange| true),
961                steps: vec![BuilderStep::Stop],
962            }
963            .span_kind_hint(),
964            SpanKindHint::Internal
965        );
966        assert_eq!(
967            BuilderStep::Split {
968                config: camel_api::splitter::SplitterConfig::new(split_body_lines()),
969                steps: vec![BuilderStep::Stop],
970            }
971            .span_kind_hint(),
972            SpanKindHint::Internal
973        );
974
975        // rc-2heu: endpoint-bearing EIP steps share the `To` scheme
976        // classification. Each covers a Client scheme, a Producer scheme,
977        // and an Internal (in-process) scheme.
978        let enrich = |uri: &str| BuilderStep::Enrich {
979            uri: uri.into(),
980            strategy: None,
981            timeout_ms: None,
982        };
983        let poll_enrich = |uri: &str| BuilderStep::PollEnrich {
984            uri: uri.into(),
985            strategy: None,
986            timeout_ms: None,
987        };
988        let wire_tap = |uri: &str| BuilderStep::WireTap { uri: uri.into() };
989
990        // Client scheme.
991        assert_eq!(enrich("http://x").span_kind_hint(), SpanKindHint::Client);
992        assert_eq!(
993            poll_enrich("http://x").span_kind_hint(),
994            SpanKindHint::Client
995        );
996        assert_eq!(wire_tap("http://x").span_kind_hint(), SpanKindHint::Client);
997        // Producer scheme.
998        assert_eq!(
999            enrich("kafka:orders").span_kind_hint(),
1000            SpanKindHint::Producer
1001        );
1002        assert_eq!(
1003            poll_enrich("kafka:orders").span_kind_hint(),
1004            SpanKindHint::Producer
1005        );
1006        assert_eq!(
1007            wire_tap("kafka:orders").span_kind_hint(),
1008            SpanKindHint::Producer
1009        );
1010        // Internal (in-process) scheme.
1011        assert_eq!(enrich("direct:y").span_kind_hint(), SpanKindHint::Internal);
1012        assert_eq!(
1013            poll_enrich("seda:q").span_kind_hint(),
1014            SpanKindHint::Internal
1015        );
1016        assert_eq!(
1017            wire_tap("direct:y").span_kind_hint(),
1018            SpanKindHint::Internal
1019        );
1020    }
1021
1022    /// Golden debug output for EVERY BuilderStep variant.
1023    ///
1024    /// Before refactoring the manual `Debug` impl to `#[derive(Debug)]`, this test
1025    /// locks the exact output of every variant so the refactor can be verified.
1026    #[test]
1027    fn golden_debug_output_all_variants() {
1028        use camel_api::declarative::LanguageExpressionDef;
1029        use camel_api::loop_eip::LoopMode;
1030        use camel_api::recipient_list::RecipientListConfig;
1031        use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
1032        use camel_api::{
1033            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
1034            FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
1035            Value,
1036        };
1037        use std::sync::Arc;
1038
1039        let expr = LanguageExpressionDef {
1040            language: "simple".into(),
1041            source: "${body}".into(),
1042        };
1043
1044        // -- group A: trivial / non-recursive ----------------------------------
1045        // derive(Debug) emits the variant without the enum name for unit/tuple variants.
1046        assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
1047        assert_eq!(
1048            format!(
1049                "{:?}",
1050                BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
1051            ),
1052            "Processor(BoxProcessor(...))"
1053        );
1054        assert_eq!(
1055            format!("{:?}", BuilderStep::To("mock:out".into())),
1056            "To(\"mock:out\")"
1057        );
1058
1059        // -- group B: variant with named fields (no sub-steps) -----------------
1060        // derive(Debug) emits ALL fields, not the redacted `..` form.
1061        assert_eq!(
1062            format!(
1063                "{:?}",
1064                BuilderStep::Log {
1065                    level: camel_processor::LogLevel::Info,
1066                    message: "hello".into(),
1067                }
1068            ),
1069            "Log { level: Info, message: \"hello\" }"
1070        );
1071
1072        assert_eq!(
1073            format!(
1074                "{:?}",
1075                BuilderStep::DeclarativeSetHeader {
1076                    key: "k".into(),
1077                    value: ValueSourceDef::Literal(Value::String("v".into())),
1078                }
1079            ),
1080            "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
1081        );
1082
1083        assert_eq!(
1084            format!(
1085                "{:?}",
1086                BuilderStep::DeclarativeSetHeaderIfAbsent {
1087                    key: "k".into(),
1088                    value: ValueSourceDef::Literal(Value::String("v".into())),
1089                }
1090            ),
1091            "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
1092        );
1093
1094        assert_eq!(
1095            format!(
1096                "{:?}",
1097                BuilderStep::DeclarativeSetBody {
1098                    value: ValueSourceDef::Literal(Value::String("v".into())),
1099                }
1100            ),
1101            "DeclarativeSetBody { value: Literal(String(\"v\")) }"
1102        );
1103
1104        assert_eq!(
1105            format!(
1106                "{:?}",
1107                BuilderStep::DeclarativeSetProperty {
1108                    key: "prop".into(),
1109                    value_source: ValueSourceDef::Literal(Value::String("v".into())),
1110                }
1111            ),
1112            "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
1113        );
1114
1115        assert_eq!(
1116            format!(
1117                "{:?}",
1118                BuilderStep::DeclarativeScript {
1119                    expression: expr.clone(),
1120                }
1121            ),
1122            "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
1123        );
1124
1125        // DeclarativeFunction needs a FunctionDefinition
1126        let func_def = FunctionDefinition {
1127            id: FunctionId("test-id".into()),
1128            runtime: "my_runtime".into(),
1129            source: "${body}".into(),
1130            timeout_ms: 5000,
1131            route_id: None,
1132            step_index: None,
1133        };
1134        assert_eq!(
1135            format!(
1136                "{:?}",
1137                BuilderStep::DeclarativeFunction {
1138                    definition: func_def,
1139                }
1140            ),
1141            "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
1142        );
1143
1144        assert_eq!(
1145            format!(
1146                "{:?}",
1147                BuilderStep::WireTap {
1148                    uri: "mock:tap".into(),
1149                }
1150            ),
1151            "WireTap { uri: \"mock:tap\" }"
1152        );
1153
1154        assert_eq!(
1155            format!(
1156                "{:?}",
1157                BuilderStep::DeclarativeLog {
1158                    level: camel_processor::LogLevel::Info,
1159                    message: ValueSourceDef::Expression(expr.clone()),
1160                }
1161            ),
1162            "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
1163        );
1164
1165        assert_eq!(
1166            format!(
1167                "{:?}",
1168                BuilderStep::Bean {
1169                    name: "myBean".into(),
1170                    method: "process".into(),
1171                }
1172            ),
1173            "Bean { name: \"myBean\", method: \"process\" }"
1174        );
1175
1176        assert_eq!(
1177            format!(
1178                "{:?}",
1179                BuilderStep::Script {
1180                    language: "js".into(),
1181                    script: "body".into(),
1182                }
1183            ),
1184            "Script { language: \"js\", script: \"body\" }"
1185        );
1186
1187        assert_eq!(
1188            format!(
1189                "{:?}",
1190                BuilderStep::Aggregate {
1191                    config: camel_api::AggregatorConfig::correlate_by("id")
1192                        .complete_when_size(1)
1193                        .build()
1194                        .unwrap(),
1195                }
1196            ),
1197            "Aggregate { config: AggregatorConfig { header_name: \"id\", completion: Single(Size(1)), correlation: HeaderName(\"id\"), strategy: CollectAll, max_buckets: Some(10000), max_bucket_size: Some(10000), bucket_ttl: Some(300s), force_completion_on_stop: false, discard_on_timeout: false, max_timeout_tasks: 1024 } }"
1198        );
1199
1200        assert_eq!(
1201            format!(
1202                "{:?}",
1203                BuilderStep::DynamicRouter {
1204                    config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
1205                        "mock:dr".into()
1206                    ))),
1207                }
1208            ),
1209            "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
1210        );
1211
1212        assert_eq!(
1213            format!(
1214                "{:?}",
1215                BuilderStep::RoutingSlip {
1216                    config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
1217                }
1218            ),
1219            "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
1220        );
1221
1222        assert_eq!(
1223            format!(
1224                "{:?}",
1225                BuilderStep::RecipientList {
1226                    config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
1227                }
1228            ),
1229            "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
1230        );
1231
1232        assert_eq!(
1233            format!(
1234                "{:?}",
1235                BuilderStep::Enrich {
1236                    uri: "mock:enrich".into(),
1237                    strategy: Some("agg".into()),
1238                    timeout_ms: Some(1000),
1239                }
1240            ),
1241            "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
1242        );
1243
1244        assert_eq!(
1245            format!(
1246                "{:?}",
1247                BuilderStep::PollEnrich {
1248                    uri: "mock:poll".into(),
1249                    strategy: None,
1250                    timeout_ms: None,
1251                }
1252            ),
1253            "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
1254        );
1255
1256        assert_eq!(
1257            format!(
1258                "{:?}",
1259                BuilderStep::Validate {
1260                    predicate: expr.clone(),
1261                }
1262            ),
1263            "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
1264        );
1265
1266        assert_eq!(
1267            format!("{:?}", BuilderStep::Sampling { period: 100 }),
1268            "Sampling { period: 100 }"
1269        );
1270
1271        assert_eq!(
1272            format!(
1273                "{:?}",
1274                BuilderStep::Resequence {
1275                    policy_config: Default::default(),
1276                }
1277            ),
1278            "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
1279        );
1280
1281        // -- group C: variants with named fields (sub-steps is Vec) ------------
1282        assert_eq!(
1283            format!(
1284                "{:?}",
1285                BuilderStep::DeclarativeFilter {
1286                    predicate: expr.clone(),
1287                    steps: vec![BuilderStep::Stop],
1288                }
1289            ),
1290            "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
1291        );
1292
1293        assert_eq!(
1294            format!(
1295                "{:?}",
1296                BuilderStep::DeclarativeSplit {
1297                    expression: expr.clone(),
1298                    aggregation: AggregationStrategy::Original,
1299                    parallel: false,
1300                    parallel_limit: Some(2),
1301                    stop_on_exception: true,
1302                    steps: vec![BuilderStep::Stop],
1303                }
1304            ),
1305            "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
1306        );
1307
1308        assert_eq!(
1309            format!(
1310                "{:?}",
1311                BuilderStep::Split {
1312                    config: camel_api::splitter::SplitterConfig::new(
1313                        camel_api::splitter::split_body_lines()
1314                    ),
1315                    steps: vec![BuilderStep::Stop],
1316                }
1317            ),
1318            "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
1319        );
1320
1321        assert_eq!(
1322            format!(
1323                "{:?}",
1324                BuilderStep::Filter {
1325                    predicate: FilterPredicate::new(|_: &Exchange| true),
1326                    steps: vec![BuilderStep::Stop],
1327                }
1328            ),
1329            "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
1330        );
1331
1332        assert_eq!(
1333            format!(
1334                "{:?}",
1335                BuilderStep::Throttle {
1336                    config: camel_api::ThrottlerConfig::new(
1337                        10,
1338                        std::time::Duration::from_millis(10)
1339                    ),
1340                    steps: vec![BuilderStep::Stop],
1341                }
1342            ),
1343            "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
1344        );
1345
1346        assert_eq!(
1347            format!(
1348                "{:?}",
1349                BuilderStep::LoadBalance {
1350                    config: camel_api::LoadBalancerConfig::round_robin(),
1351                    steps: vec![BuilderStep::To("mock:l1".into())],
1352                }
1353            ),
1354            "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
1355        );
1356
1357        assert_eq!(
1358            format!(
1359                "{:?}",
1360                BuilderStep::Delay {
1361                    config: camel_api::DelayConfig::new(500),
1362                }
1363            ),
1364            "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
1365        );
1366
1367        // -- group D: Choice / DeclarativeChoice --------------------------------
1368        // derive(Debug) emits full field enumeration; WhenStep Debug is a
1369        // full struct listing; nested BuilderStep::Stop stays as `Stop`.
1370        assert_eq!(
1371            format!(
1372                "{:?}",
1373                BuilderStep::Choice {
1374                    whens: vec![WhenStep {
1375                        predicate: FilterPredicate::new(|_: &Exchange| true),
1376                        steps: vec![BuilderStep::To("mock:a".into())],
1377                    }],
1378                    otherwise: None,
1379                }
1380            ),
1381            "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
1382        );
1383
1384        assert_eq!(
1385            format!(
1386                "{:?}",
1387                BuilderStep::DeclarativeChoice {
1388                    whens: vec![DeclarativeWhenStep {
1389                        predicate: expr.clone(),
1390                        steps: vec![BuilderStep::Stop],
1391                    }],
1392                    otherwise: Some(vec![BuilderStep::Stop]),
1393                }
1394            ),
1395            "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
1396        );
1397
1398        // -- group E: Multicast -------------------------------------------------
1399        assert_eq!(
1400            format!(
1401                "{:?}",
1402                BuilderStep::Multicast {
1403                    steps: vec![BuilderStep::To("direct:a".into())],
1404                    config: MulticastConfig::new(),
1405                }
1406            ),
1407            "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
1408        );
1409
1410        // -- group F: DeclarativeDynamicRouter / DeclarativeRoutingSlip --------
1411        assert_eq!(
1412            format!(
1413                "{:?}",
1414                BuilderStep::DeclarativeDynamicRouter {
1415                    expression: expr.clone(),
1416                    uri_delimiter: ",".into(),
1417                    cache_size: 1000,
1418                    ignore_invalid_endpoints: false,
1419                    max_iterations: 1000,
1420                }
1421            ),
1422            "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
1423        );
1424
1425        assert_eq!(
1426            format!(
1427                "{:?}",
1428                BuilderStep::DeclarativeRoutingSlip {
1429                    expression: expr.clone(),
1430                    uri_delimiter: ",".into(),
1431                    cache_size: 1000,
1432                    ignore_invalid_endpoints: false,
1433                }
1434            ),
1435            "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
1436        );
1437
1438        // -- group G: DeclarativeRecipientList ---------------------------------
1439        assert_eq!(
1440            format!(
1441                "{:?}",
1442                BuilderStep::DeclarativeRecipientList {
1443                    expression: expr.clone(),
1444                    delimiter: ",".into(),
1445                    parallel: false,
1446                    parallel_limit: None,
1447                    stop_on_exception: false,
1448                    aggregation: "original".into(),
1449                }
1450            ),
1451            "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
1452        );
1453
1454        // -- group H: Loop / DeclarativeLoop -----------------------------------
1455        assert_eq!(
1456            format!(
1457                "{:?}",
1458                BuilderStep::Loop {
1459                    config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
1460                    steps: vec![],
1461                }
1462            ),
1463            "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
1464        );
1465
1466        assert_eq!(
1467            format!(
1468                "{:?}",
1469                BuilderStep::DeclarativeLoop {
1470                    count: Some(5),
1471                    while_predicate: None,
1472                    steps: vec![],
1473                    max_iterations: Some(100),
1474                }
1475            ),
1476            "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
1477        );
1478
1479        // -- group I: ClaimCheck -----------------------------------------------
1480        assert_eq!(
1481            format!(
1482                "{:?}",
1483                BuilderStep::ClaimCheck {
1484                    repository: "myRepo".into(),
1485                    operation: "checkout".into(),
1486                    key: expr.clone(),
1487                    filter: None,
1488                }
1489            ),
1490            "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
1491        );
1492
1493        // -- group J: Sort -----------------------------------------------------
1494        assert_eq!(
1495            format!(
1496                "{:?}",
1497                BuilderStep::Sort {
1498                    expression: expr.clone(),
1499                    reverse: false,
1500                }
1501            ),
1502            "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
1503        );
1504
1505        // -- group K: IdempotentConsumer ---------------------------------------
1506        assert_eq!(
1507            format!(
1508                "{:?}",
1509                BuilderStep::IdempotentConsumer {
1510                    repository: "myRepo".into(),
1511                    expression: expr.clone(),
1512                    steps: vec![],
1513                    eager: true,
1514                    remove_on_failure: false,
1515                }
1516            ),
1517            "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
1518        );
1519
1520        // -- group L: DeclarativeDoTry -----------------------------------------
1521        assert_eq!(
1522            format!(
1523                "{:?}",
1524                BuilderStep::DeclarativeDoTry {
1525                    try_steps: vec![BuilderStep::Stop],
1526                    catch: vec![],
1527                    finally: None,
1528                }
1529            ),
1530            "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1531        );
1532
1533        // -- group M: DeclarativeStreamSplit -----------------------------------
1534        assert_eq!(
1535            format!(
1536                "{:?}",
1537                BuilderStep::DeclarativeStreamSplit {
1538                    stream_config: StreamSplitConfig {
1539                        format: StreamSplitFormat::Ndjson,
1540                        max_record_bytes: 1024 * 1024,
1541                        batch_size: 1,
1542                        chunk_size: None,
1543                        include_origin: true,
1544                    },
1545                    aggregation: AggregationStrategy::Original,
1546                    stop_on_exception: true,
1547                    steps: vec![BuilderStep::Stop],
1548                }
1549            ),
1550            "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] }"
1551        );
1552    }
1553
1554    #[test]
1555    fn test_builder_step_multicast_variant() {
1556        use camel_api::MulticastConfig;
1557
1558        let step = BuilderStep::Multicast {
1559            steps: vec![BuilderStep::To("direct:a".into())],
1560            config: MulticastConfig::new(),
1561        };
1562
1563        assert!(matches!(step, BuilderStep::Multicast { .. }));
1564    }
1565
1566    #[test]
1567    fn test_route_definition_defaults() {
1568        let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1569        assert_eq!(def.route_id(), "test-route");
1570        assert!(def.auto_startup());
1571        assert_eq!(def.startup_order(), 1000);
1572    }
1573
1574    #[test]
1575    fn test_route_definition_builders() {
1576        let def = RouteDefinition::new("direct:test", vec![])
1577            .with_route_id("my-route")
1578            .with_auto_startup(false)
1579            .with_startup_order(50);
1580        assert_eq!(def.route_id(), "my-route");
1581        assert!(!def.auto_startup());
1582        assert_eq!(def.startup_order(), 50);
1583    }
1584
1585    #[test]
1586    fn test_route_definition_accessors_cover_core_fields() {
1587        let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1588            .with_route_id("accessor-route");
1589
1590        assert_eq!(def.from_uri(), "direct:in");
1591        assert_eq!(def.steps().len(), 1);
1592        assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1593    }
1594
1595    #[test]
1596    fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1597        use camel_api::circuit_breaker::CircuitBreakerConfig;
1598        use camel_api::error_handler::ErrorHandlerConfig;
1599        use camel_component_api::ConcurrencyModel;
1600
1601        let def = RouteDefinition::new("direct:test", vec![])
1602            .with_route_id("eh-route")
1603            .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1604            .with_circuit_breaker(CircuitBreakerConfig::new())
1605            .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1606
1607        let eh = def
1608            .error_handler_config()
1609            .expect("error handler should be set");
1610        assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1611        assert!(def.circuit_breaker_config().is_some());
1612        assert!(matches!(
1613            def.concurrency_override(),
1614            Some(ConcurrencyModel::Concurrent { max: Some(4) })
1615        ));
1616    }
1617
1618    #[test]
1619    fn test_builder_step_debug_covers_many_variants() {
1620        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1621        use camel_api::{
1622            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1623            OpaqueProcessor, RoutingSlipConfig, Value,
1624        };
1625        use std::sync::Arc;
1626
1627        let expr = LanguageExpressionDef {
1628            language: "simple".into(),
1629            source: "${body}".into(),
1630        };
1631
1632        let steps = vec![
1633            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1634            BuilderStep::To("mock:out".into()),
1635            BuilderStep::Stop,
1636            BuilderStep::Log {
1637                level: camel_processor::LogLevel::Info,
1638                message: "hello".into(),
1639            },
1640            BuilderStep::DeclarativeSetHeader {
1641                key: "k".into(),
1642                value: ValueSourceDef::Literal(Value::String("v".into())),
1643            },
1644            BuilderStep::DeclarativeSetBody {
1645                value: ValueSourceDef::Expression(expr.clone()),
1646            },
1647            BuilderStep::DeclarativeFilter {
1648                predicate: expr.clone(),
1649                steps: vec![BuilderStep::Stop],
1650            },
1651            BuilderStep::DeclarativeChoice {
1652                whens: vec![DeclarativeWhenStep {
1653                    predicate: expr.clone(),
1654                    steps: vec![BuilderStep::Stop],
1655                }],
1656                otherwise: Some(vec![BuilderStep::Stop]),
1657            },
1658            BuilderStep::DeclarativeScript {
1659                expression: expr.clone(),
1660            },
1661            BuilderStep::DeclarativeSplit {
1662                expression: expr.clone(),
1663                aggregation: AggregationStrategy::Original,
1664                parallel: false,
1665                parallel_limit: Some(2),
1666                stop_on_exception: true,
1667                steps: vec![BuilderStep::Stop],
1668            },
1669            BuilderStep::Split {
1670                config: SplitterConfig::new(split_body_lines()),
1671                steps: vec![BuilderStep::Stop],
1672            },
1673            BuilderStep::Aggregate {
1674                config: camel_api::AggregatorConfig::correlate_by("id")
1675                    .complete_when_size(1)
1676                    .build()
1677                    .unwrap(),
1678            },
1679            BuilderStep::Filter {
1680                predicate: FilterPredicate::new(|_: &Exchange| true),
1681                steps: vec![BuilderStep::Stop],
1682            },
1683            BuilderStep::WireTap {
1684                uri: "mock:tap".into(),
1685            },
1686            BuilderStep::DeclarativeLog {
1687                level: camel_processor::LogLevel::Info,
1688                message: ValueSourceDef::Expression(expr.clone()),
1689            },
1690            BuilderStep::Bean {
1691                name: "bean".into(),
1692                method: "call".into(),
1693            },
1694            BuilderStep::Script {
1695                language: "rhai".into(),
1696                script: "body".into(),
1697            },
1698            BuilderStep::Throttle {
1699                config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1700                steps: vec![BuilderStep::Stop],
1701            },
1702            BuilderStep::LoadBalance {
1703                config: camel_api::LoadBalancerConfig::round_robin(),
1704                steps: vec![BuilderStep::To("mock:l1".into())],
1705            },
1706            BuilderStep::DynamicRouter {
1707                config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1708            },
1709            BuilderStep::RoutingSlip {
1710                config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1711            },
1712        ];
1713
1714        for step in steps {
1715            let dbg = format!("{step:?}");
1716            assert!(!dbg.is_empty());
1717        }
1718    }
1719
1720    #[test]
1721    fn test_route_definition_to_info_preserves_metadata() {
1722        let info = RouteDefinition::new("direct:test", vec![])
1723            .with_route_id("meta-route")
1724            .with_auto_startup(false)
1725            .with_startup_order(7)
1726            .to_info();
1727
1728        assert_eq!(info.route_id(), "meta-route");
1729        assert!(!info.auto_startup());
1730        assert_eq!(info.startup_order(), 7);
1731    }
1732
1733    #[test]
1734    fn test_choice_builder_step_debug() {
1735        use camel_api::FilterPredicate;
1736
1737        fn always_true(_: &camel_api::Exchange) -> bool {
1738            true
1739        }
1740
1741        let step = BuilderStep::Choice {
1742            whens: vec![WhenStep {
1743                predicate: FilterPredicate::new(always_true),
1744                steps: vec![BuilderStep::To("mock:a".into())],
1745            }],
1746            otherwise: None,
1747        };
1748        let debug = format!("{step:?}");
1749        assert!(debug.contains("Choice"));
1750    }
1751
1752    #[test]
1753    fn test_route_definition_unit_of_work() {
1754        use camel_api::UnitOfWorkConfig;
1755        let config = UnitOfWorkConfig {
1756            on_complete: Some("log:complete".into()),
1757            on_failure: Some("log:failed".into()),
1758        };
1759        let def = RouteDefinition::new("direct:test", vec![])
1760            .with_route_id("uow-test")
1761            .with_unit_of_work(config.clone());
1762        assert_eq!(
1763            def.unit_of_work_config().unwrap().on_complete.as_deref(),
1764            Some("log:complete")
1765        );
1766        assert_eq!(
1767            def.unit_of_work_config().unwrap().on_failure.as_deref(),
1768            Some("log:failed")
1769        );
1770
1771        let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1772        assert!(def_no_uow.unit_of_work_config().is_none());
1773    }
1774
1775    #[test]
1776    fn test_route_definition_security_policy_accessor() {
1777        use async_trait::async_trait;
1778        use camel_api::CamelError;
1779        use camel_api::Exchange;
1780        use camel_api::security_policy::{
1781            AuthContext, AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1782        };
1783
1784        struct StubPolicy;
1785        #[async_trait]
1786        impl SecurityPolicy for StubPolicy {
1787            async fn evaluate(
1788                &self,
1789                _exchange: &mut Exchange,
1790                _auth: &AuthContext<'_>,
1791            ) -> Result<AuthorizationDecision, CamelError> {
1792                Ok(AuthorizationDecision::Granted {
1793                    principal: Principal {
1794                        subject: "test".into(),
1795                        issuer: "test".into(),
1796                        audience: vec![],
1797                        scopes: vec![],
1798                        roles: vec![],
1799                        claims: serde_json::Value::Null,
1800                    },
1801                })
1802            }
1803        }
1804
1805        let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1806        assert!(def_no_sp.security_policy_config().is_none());
1807
1808        let def = RouteDefinition::new("direct:test", vec![])
1809            .with_route_id("sp-test")
1810            .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1811        assert!(def.security_policy_config().is_some());
1812    }
1813
1814    #[test]
1815    fn test_route_definition_security_authenticator_accessor() {
1816        use camel_api::security_policy::Principal;
1817
1818        struct TestAuth;
1819        #[async_trait::async_trait]
1820        impl TokenAuthenticator for TestAuth {
1821            async fn authenticate_bearer(
1822                &self,
1823                _token: &str,
1824            ) -> Result<Principal, camel_api::CamelError> {
1825                Ok(Principal {
1826                    subject: "test".into(),
1827                    issuer: "test".into(),
1828                    audience: vec![],
1829                    scopes: vec![],
1830                    roles: vec![],
1831                    claims: serde_json::Value::Null,
1832                })
1833            }
1834        }
1835
1836        let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1837        assert!(def_no_auth.security_authenticator().is_none());
1838
1839        let auth = Arc::new(TestAuth);
1840        let def = RouteDefinition::new("direct:test".to_string(), vec![])
1841            .with_security_authenticator(auth);
1842        assert!(def.security_authenticator().is_some());
1843    }
1844
1845    #[test]
1846    fn test_map_steps_swaps_steps_and_preserves_other_fields() {
1847        let original = RouteDefinition::new(
1848            "direct:test".to_string(),
1849            vec![BuilderStep::To("mock:a".into()), BuilderStep::Stop],
1850        )
1851        .with_route_id("my-route");
1852
1853        let mapped = original.map_steps(|steps| {
1854            let mut out = Vec::with_capacity(steps.len() + 1);
1855            out.push(BuilderStep::To("mock:prefix".into()));
1856            out.extend(steps);
1857            out
1858        });
1859
1860        // Steps were transformed.
1861        assert_eq!(mapped.steps().len(), 3);
1862        assert!(matches!(mapped.steps()[0], BuilderStep::To(ref s) if s == "mock:prefix"));
1863        assert!(matches!(mapped.steps()[1], BuilderStep::To(ref s) if s == "mock:a"));
1864        // Other fields preserved.
1865        assert_eq!(mapped.route_id(), "my-route");
1866    }
1867
1868    #[test]
1869    fn circuit_breaker_fallback_accessor_returns_steps() {
1870        let def = RouteDefinition::new("direct:start", vec![])
1871            .with_circuit_breaker_fallback(vec![BuilderStep::To("mock:out".into())]);
1872        assert_eq!(def.circuit_breaker_fallback().len(), 1);
1873    }
1874}
camel_core/lifecycle/application/route_definition.rs

camel_core/lifecycle/application/
route_definition.rs
