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    /// `To` classifies the AUTHORED URI scheme — the text before the first
408    /// `:`, compared with `eq_ignore_ascii_case`, no endpoint resolution —
409    /// so a `SkipTo` interception never rewrites the kind. Messaging broker
410    /// schemes map to `Producer` (async, one-way send) and synchronous
411    /// outbound protocols to `Client`; every other scheme, scheme-less URIs,
412    /// and every non-`To` variant stay `Internal` route processing. The
413    /// catch-all sits after the `To` arm so it can never swallow it.
414    pub(crate) fn span_kind_hint(&self) -> SpanKindHint {
415        /// Broker-style destinations: an async, one-way send.
416        const PRODUCER_SCHEMES: [&str; 5] = ["kafka", "jms", "activemq", "artemis", "mqtt"];
417        /// Synchronous outbound request/response protocols.
418        const CLIENT_SCHEMES: [&str; 12] = [
419            "http",
420            "https",
421            "grpc",
422            "grpcs",
423            "ws",
424            "redis",
425            "opensearch",
426            "sql",
427            "surrealdb",
428            "cxf",
429            "llm",
430            "mcp",
431        ];
432
433        match self {
434            Self::To(uri) => {
435                // Scheme of the authored URI. Scheme-less URIs match no
436                // known scheme and stay Internal — the kind is never guessed.
437                let scheme = uri.split(':').next().unwrap_or_default();
438                if PRODUCER_SCHEMES
439                    .iter()
440                    .any(|s| s.eq_ignore_ascii_case(scheme))
441                {
442                    SpanKindHint::Producer
443                } else if CLIENT_SCHEMES
444                    .iter()
445                    .any(|s| s.eq_ignore_ascii_case(scheme))
446                {
447                    SpanKindHint::Client
448                } else {
449                    SpanKindHint::Internal
450                }
451            }
452            _ => SpanKindHint::Internal,
453        }
454    }
455}
456
457/// An unresolved route definition. "to" URIs have not been resolved to producers yet.
458pub struct RouteDefinition {
459    pub(crate) from_uri: String,
460    pub(crate) steps: Vec<BuilderStep>,
461    /// Optional per-route error handler config. Takes precedence over the global one.
462    pub(crate) error_handler: Option<ErrorHandlerConfig>,
463    /// Optional circuit breaker config. Applied between error handler and step pipeline.
464    pub(crate) circuit_breaker: Option<CircuitBreakerConfig>,
465    /// Circuit breaker fallback sub-pipeline, as UNRESOLVED steps.
466    ///
467    /// Sibling of [`RouteDefinition::circuit_breaker`]: the DSL layers thread
468    /// `Vec<BuilderStep>` here (same rule as `cache_peek_stale.on_miss`) and
469    /// camel-core compiles it via the `StepCompilerRegistry` when the route is
470    /// compiled. The resolved `CircuitBreakerConfig.fallback` (`BoxProcessor`)
471    /// stays `None` until that compile — the DSL never constructs processors.
472    pub(crate) circuit_breaker_fallback: Vec<BuilderStep>,
473    pub(crate) security_policy: Option<SecurityPolicyConfig>,
474    /// Optional token authenticator for validating JWT/OAuth tokens.
475    pub(crate) security_authenticator: Option<Arc<dyn TokenAuthenticator>>,
476    /// Named authentication providers for this route, injected into the
477    /// consumer `SecurityContext` so Phase-2 transports can resolve them.
478    pub(crate) provider_registry: Option<Arc<camel_auth::ProviderRegistry>>,
479    /// Declared provider name for plan compilation (`security_policy.provider`
480    /// in the DSL). Resolved against the provider registry at staging time;
481    /// a missing name fails compilation instead of downgrading to `Public`.
482    pub(crate) security_provider: Option<String>,
483    /// Route-level audience override for plan compilation. When present the
484    /// compiled `RouteSecurityPlan.audience_binding` uses these audiences
485    /// (issuers still come from the provider); when absent the resolved
486    /// provider's binding is copied verbatim.
487    pub(crate) security_audiences: Option<Vec<String>>,
488    /// Optional Unit of Work config for in-flight tracking and completion hooks.
489    pub(crate) unit_of_work: Option<UnitOfWorkConfig>,
490    /// User override for the consumer's concurrency model. `None` means
491    /// "use whatever the consumer declares".
492    pub(crate) concurrency: Option<ConcurrencyModel>,
493    /// Unique identifier for this route. Required.
494    pub(crate) route_id: String,
495    /// Whether this route should start automatically when the context starts.
496    pub(crate) auto_startup: bool,
497    /// Order in which routes are started. Lower values start first.
498    pub(crate) startup_order: i32,
499    pub(crate) source_hash: Option<u64>,
500}
501
502impl RouteDefinition {
503    /// Create a new route definition with the required route ID.
504    pub fn new(from_uri: impl Into<String>, steps: Vec<BuilderStep>) -> Self {
505        Self {
506            from_uri: from_uri.into(),
507            steps,
508            error_handler: None,
509            circuit_breaker: None,
510            circuit_breaker_fallback: Vec::new(),
511            security_policy: None,
512            security_authenticator: None,
513            provider_registry: None,
514            security_provider: None,
515            security_audiences: None,
516            unit_of_work: None,
517            concurrency: None,
518            route_id: String::new(), // Will be set by with_route_id()
519            auto_startup: true,
520            startup_order: 1000,
521            source_hash: None,
522        }
523    }
524
525    /// The source endpoint URI.
526    pub fn from_uri(&self) -> &str {
527        &self.from_uri
528    }
529
530    /// The steps in this route definition.
531    pub fn steps(&self) -> &[BuilderStep] {
532        &self.steps
533    }
534
535    pub fn circuit_breaker_fallback(&self) -> &[BuilderStep] {
536        &self.circuit_breaker_fallback
537    }
538
539    /// Transform the step list, consuming and returning self.
540    /// Used for post-parse instrumentation (e.g. benchmark timing injection
541    /// around `To` steps). The field stays private; callers rebuild via this
542    /// consuming method rather than holding a `&mut` alias.
543    pub fn map_steps(mut self, f: impl FnOnce(Vec<BuilderStep>) -> Vec<BuilderStep>) -> Self {
544        self.steps = f(self.steps);
545        self
546    }
547
548    /// Set a per-route error handler, overriding the global one.
549    pub fn with_error_handler(mut self, config: ErrorHandlerConfig) -> Self {
550        self.error_handler = Some(config);
551        self
552    }
553
554    /// Get the route-level error handler config, if set.
555    pub fn error_handler_config(&self) -> Option<&ErrorHandlerConfig> {
556        self.error_handler.as_ref()
557    }
558
559    /// Set a circuit breaker for this route.
560    pub fn with_circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
561        self.circuit_breaker = Some(config);
562        self
563    }
564
565    /// Set the circuit breaker fallback sub-pipeline (unresolved steps).
566    ///
567    /// Mirror of [`RouteDefinition::with_circuit_breaker`]: the steps are
568    /// compiled by camel-core at route-compile time (registry monopoly) and
569    /// attached to the resolved `CircuitBreakerConfig.fallback`. Empty when no
570    /// CB is configured or the CB has no fallback.
571    pub fn with_circuit_breaker_fallback(mut self, steps: Vec<BuilderStep>) -> Self {
572        self.circuit_breaker_fallback = steps;
573        self
574    }
575
576    /// Set a security policy for this route.
577    pub fn with_security_policy(mut self, config: SecurityPolicyConfig) -> Self {
578        self.security_policy = Some(config);
579        self
580    }
581
582    /// Set a token authenticator for this route.
583    pub fn with_security_authenticator(
584        mut self,
585        authenticator: Arc<dyn TokenAuthenticator>,
586    ) -> Self {
587        self.security_authenticator = Some(authenticator);
588        self
589    }
590
591    /// Set the named authentication providers for this route.
592    ///
593    /// The registry is injected into the consumer `SecurityContext` so
594    /// Phase-2 transports can resolve providers without holding their own
595    /// authenticator.
596    pub fn with_provider_registry(mut self, registry: Arc<camel_auth::ProviderRegistry>) -> Self {
597        self.provider_registry = Some(registry);
598        self
599    }
600
601    /// Set the declared provider name used by plan compilation.
602    pub fn with_security_provider(mut self, name: impl Into<String>) -> Self {
603        self.security_provider = Some(name.into());
604        self
605    }
606
607    /// Set the route-level audience override used by plan compilation.
608    pub fn with_security_audiences(mut self, audiences: Vec<String>) -> Self {
609        self.security_audiences = Some(audiences);
610        self
611    }
612
613    /// The declared provider name, if any.
614    pub fn security_provider(&self) -> Option<&str> {
615        self.security_provider.as_deref()
616    }
617
618    /// The route-level audience override, if any.
619    pub fn security_audiences(&self) -> Option<&[String]> {
620        self.security_audiences.as_deref()
621    }
622
623    /// Set a unit of work config for this route.
624    pub fn with_unit_of_work(mut self, config: UnitOfWorkConfig) -> Self {
625        self.unit_of_work = Some(config);
626        self
627    }
628
629    /// Get the unit of work config, if set.
630    pub fn unit_of_work_config(&self) -> Option<&UnitOfWorkConfig> {
631        self.unit_of_work.as_ref()
632    }
633
634    /// Get the circuit breaker config, if set.
635    pub fn circuit_breaker_config(&self) -> Option<&CircuitBreakerConfig> {
636        self.circuit_breaker.as_ref()
637    }
638
639    pub fn security_policy_config(&self) -> Option<&SecurityPolicyConfig> {
640        self.security_policy.as_ref()
641    }
642
643    pub fn security_authenticator(&self) -> Option<&Arc<dyn TokenAuthenticator>> {
644        self.security_authenticator.as_ref()
645    }
646
647    /// User-specified concurrency override, if any.
648    pub fn concurrency_override(&self) -> Option<&ConcurrencyModel> {
649        self.concurrency.as_ref()
650    }
651
652    /// Override the consumer's concurrency model for this route.
653    pub fn with_concurrency(mut self, model: ConcurrencyModel) -> Self {
654        self.concurrency = Some(model);
655        self
656    }
657
658    /// Get the route ID.
659    pub fn route_id(&self) -> &str {
660        &self.route_id
661    }
662
663    /// Whether this route should start automatically when the context starts.
664    pub fn auto_startup(&self) -> bool {
665        self.auto_startup
666    }
667
668    /// Order in which routes are started. Lower values start first.
669    pub fn startup_order(&self) -> i32 {
670        self.startup_order
671    }
672
673    /// Set a unique identifier for this route.
674    pub fn with_route_id(mut self, id: impl Into<String>) -> Self {
675        self.route_id = id.into();
676        self
677    }
678
679    /// Set whether this route should start automatically.
680    pub fn with_auto_startup(mut self, auto: bool) -> Self {
681        self.auto_startup = auto;
682        self
683    }
684
685    /// Set the startup order. Lower values start first.
686    pub fn with_startup_order(mut self, order: i32) -> Self {
687        self.startup_order = order;
688        self
689    }
690
691    pub fn with_source_hash(mut self, hash: u64) -> Self {
692        self.source_hash = Some(hash);
693        self
694    }
695
696    pub fn source_hash(&self) -> Option<u64> {
697        self.source_hash
698    }
699
700    /// Extract the metadata fields needed for introspection.
701    /// This is used by RouteController to store route info without the non-Sync steps.
702    pub fn to_info(&self) -> RouteDefinitionInfo {
703        RouteDefinitionInfo {
704            route_id: self.route_id.clone(),
705            auto_startup: self.auto_startup,
706            startup_order: self.startup_order,
707            source_hash: self.source_hash,
708        }
709    }
710}
711
712/// Minimal route definition metadata for introspection.
713///
714/// This struct contains only the metadata fields from [`RouteDefinition`]
715/// that are needed for route lifecycle management, without the `steps` field
716/// (which contains non-Sync types and cannot be stored in a Sync struct).
717#[derive(Clone)]
718pub struct RouteDefinitionInfo {
719    route_id: String,
720    auto_startup: bool,
721    startup_order: i32,
722    pub(crate) source_hash: Option<u64>,
723}
724
725impl RouteDefinitionInfo {
726    /// Get the route ID.
727    pub fn route_id(&self) -> &str {
728        &self.route_id
729    }
730
731    /// Whether this route should start automatically when the context starts.
732    pub fn auto_startup(&self) -> bool {
733        self.auto_startup
734    }
735
736    /// Order in which routes are started. Lower values start first.
737    pub fn startup_order(&self) -> i32 {
738        self.startup_order
739    }
740
741    pub fn source_hash(&self) -> Option<u64> {
742        self.source_hash
743    }
744}
745
746#[cfg(test)]
747mod tests {
748    use super::*;
749
750    /// Task 1.2 (span-name-enrichment): `BuilderStep::span_label` mapping.
751    ///
752    /// `To` labels carry the authored URI scheme (before the first `:`, no
753    /// interception resolution); EIP variants map to kebab-case EIP names
754    /// (declarative twins share the label of their programmatic sibling);
755    /// anonymous variants (opaque processors) carry no label.
756    #[test]
757    fn builder_step_span_label_mapping() {
758        use camel_api::declarative::LanguageExpressionDef;
759        use camel_api::splitter::AggregationStrategy;
760        use camel_api::{BoxProcessor, IdentityProcessor, OpaqueProcessor};
761
762        let expr = LanguageExpressionDef {
763            language: "simple".into(),
764            source: "${body}".into(),
765        };
766
767        // To: scheme is the authored URI before the first ':'.
768        assert_eq!(
769            BuilderStep::To("direct:tree-sub".into())
770                .span_label()
771                .as_deref(),
772            Some("to:direct")
773        );
774        assert_eq!(
775            BuilderStep::To("http://api.example/x".into())
776                .span_label()
777                .as_deref(),
778            Some("to:http")
779        );
780        // Scheme-less URI: no label — never leak the full text.
781        assert_eq!(BuilderStep::To("garbage".into()).span_label(), None);
782
783        // EIP variants → kebab-case EIP name.
784        assert_eq!(
785            BuilderStep::Log {
786                level: camel_processor::LogLevel::Info,
787                message: "m".into(),
788            }
789            .span_label()
790            .as_deref(),
791            Some("log")
792        );
793        assert_eq!(
794            BuilderStep::Split {
795                config: camel_api::splitter::SplitterConfig::new(
796                    camel_api::splitter::split_body_lines()
797                ),
798                steps: vec![BuilderStep::Stop],
799            }
800            .span_label()
801            .as_deref(),
802            Some("split")
803        );
804        assert_eq!(
805            BuilderStep::DeclarativeSplit {
806                expression: expr.clone(),
807                aggregation: AggregationStrategy::Original,
808                parallel: false,
809                parallel_limit: None,
810                stop_on_exception: true,
811                steps: vec![BuilderStep::Stop],
812            }
813            .span_label()
814            .as_deref(),
815            Some("split")
816        );
817        assert_eq!(
818            BuilderStep::DeclarativeStreamSplit {
819                stream_config: camel_api::StreamSplitConfig::default(),
820                aggregation: AggregationStrategy::Original,
821                stop_on_exception: true,
822                steps: vec![BuilderStep::Stop],
823            }
824            .span_label()
825            .as_deref(),
826            Some("split")
827        );
828        assert_eq!(BuilderStep::Stop.span_label(), None);
829
830        // Anonymous variants → None.
831        assert_eq!(
832            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
833                .span_label(),
834            None
835        );
836    }
837
838    /// Task 1.2 (span-kind-hint): `BuilderStep::span_kind_hint` mapping.
839    ///
840    /// `To` classifies the AUTHORED URI scheme (text before the first `:`,
841    /// compared case-insensitively, no endpoint resolution): messaging
842    /// brokers map to `Producer`, synchronous outbound protocols to
843    /// `Client`. Every other scheme, scheme-less URIs, and every non-`To`
844    /// variant stay `Internal` route processing.
845    #[test]
846    fn builder_step_span_kind_hint_mapping() {
847        use camel_api::splitter::split_body_lines;
848        use camel_api::{Exchange, FilterPredicate, SpanKindHint};
849
850        let kind = |uri: &str| BuilderStep::To(uri.into()).span_kind_hint();
851
852        // Messaging brokers → Producer.
853        assert_eq!(kind("kafka:orders"), SpanKindHint::Producer);
854        assert_eq!(kind("jms:q"), SpanKindHint::Producer);
855        assert_eq!(kind("activemq:q"), SpanKindHint::Producer);
856        assert_eq!(kind("artemis:q"), SpanKindHint::Producer);
857        assert_eq!(kind("mqtt:t"), SpanKindHint::Producer);
858        // Scheme comparison is case-insensitive.
859        assert_eq!(kind("KAFKA:orders"), SpanKindHint::Producer);
860
861        // Synchronous outbound protocols → Client.
862        assert_eq!(kind("http://x"), SpanKindHint::Client);
863        assert_eq!(kind("https://x"), SpanKindHint::Client);
864        assert_eq!(kind("grpc://x"), SpanKindHint::Client);
865        assert_eq!(kind("grpcs://x"), SpanKindHint::Client);
866        assert_eq!(kind("ws://x"), SpanKindHint::Client);
867        assert_eq!(kind("redis://x"), SpanKindHint::Client);
868        assert_eq!(kind("opensearch://x"), SpanKindHint::Client);
869        assert_eq!(kind("sql:db"), SpanKindHint::Client);
870        assert_eq!(kind("surrealdb://x"), SpanKindHint::Client);
871        assert_eq!(kind("cxf://x"), SpanKindHint::Client);
872        assert_eq!(kind("llm://x"), SpanKindHint::Client);
873        assert_eq!(kind("mcp://x"), SpanKindHint::Client);
874
875        // In-process and unknown schemes → Internal.
876        assert_eq!(kind("direct:y"), SpanKindHint::Internal);
877        assert_eq!(kind("timer:z"), SpanKindHint::Internal);
878        // Scheme-less URI: the kind is never guessed.
879        assert_eq!(kind("garbage"), SpanKindHint::Internal);
880
881        // Non-`To` variants are internal route processing.
882        assert_eq!(
883            BuilderStep::Log {
884                level: camel_processor::LogLevel::Info,
885                message: "m".into(),
886            }
887            .span_kind_hint(),
888            SpanKindHint::Internal
889        );
890        assert_eq!(
891            BuilderStep::Filter {
892                predicate: FilterPredicate::new(|_: &Exchange| true),
893                steps: vec![BuilderStep::Stop],
894            }
895            .span_kind_hint(),
896            SpanKindHint::Internal
897        );
898        assert_eq!(
899            BuilderStep::Split {
900                config: camel_api::splitter::SplitterConfig::new(split_body_lines()),
901                steps: vec![BuilderStep::Stop],
902            }
903            .span_kind_hint(),
904            SpanKindHint::Internal
905        );
906    }
907
908    /// Golden debug output for EVERY BuilderStep variant.
909    ///
910    /// Before refactoring the manual `Debug` impl to `#[derive(Debug)]`, this test
911    /// locks the exact output of every variant so the refactor can be verified.
912    #[test]
913    fn golden_debug_output_all_variants() {
914        use camel_api::declarative::LanguageExpressionDef;
915        use camel_api::loop_eip::LoopMode;
916        use camel_api::recipient_list::RecipientListConfig;
917        use camel_api::splitter::{AggregationStrategy, StreamSplitConfig, StreamSplitFormat};
918        use camel_api::{
919            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, FunctionDefinition,
920            FunctionId, IdentityProcessor, MulticastConfig, OpaqueProcessor, RoutingSlipConfig,
921            Value,
922        };
923        use std::sync::Arc;
924
925        let expr = LanguageExpressionDef {
926            language: "simple".into(),
927            source: "${body}".into(),
928        };
929
930        // -- group A: trivial / non-recursive ----------------------------------
931        // derive(Debug) emits the variant without the enum name for unit/tuple variants.
932        assert_eq!(format!("{:?}", BuilderStep::Stop), "Stop");
933        assert_eq!(
934            format!(
935                "{:?}",
936                BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor)))
937            ),
938            "Processor(BoxProcessor(...))"
939        );
940        assert_eq!(
941            format!("{:?}", BuilderStep::To("mock:out".into())),
942            "To(\"mock:out\")"
943        );
944
945        // -- group B: variant with named fields (no sub-steps) -----------------
946        // derive(Debug) emits ALL fields, not the redacted `..` form.
947        assert_eq!(
948            format!(
949                "{:?}",
950                BuilderStep::Log {
951                    level: camel_processor::LogLevel::Info,
952                    message: "hello".into(),
953                }
954            ),
955            "Log { level: Info, message: \"hello\" }"
956        );
957
958        assert_eq!(
959            format!(
960                "{:?}",
961                BuilderStep::DeclarativeSetHeader {
962                    key: "k".into(),
963                    value: ValueSourceDef::Literal(Value::String("v".into())),
964                }
965            ),
966            "DeclarativeSetHeader { key: \"k\", value: Literal(String(\"v\")) }"
967        );
968
969        assert_eq!(
970            format!(
971                "{:?}",
972                BuilderStep::DeclarativeSetHeaderIfAbsent {
973                    key: "k".into(),
974                    value: ValueSourceDef::Literal(Value::String("v".into())),
975                }
976            ),
977            "DeclarativeSetHeaderIfAbsent { key: \"k\", value: Literal(String(\"v\")) }"
978        );
979
980        assert_eq!(
981            format!(
982                "{:?}",
983                BuilderStep::DeclarativeSetBody {
984                    value: ValueSourceDef::Literal(Value::String("v".into())),
985                }
986            ),
987            "DeclarativeSetBody { value: Literal(String(\"v\")) }"
988        );
989
990        assert_eq!(
991            format!(
992                "{:?}",
993                BuilderStep::DeclarativeSetProperty {
994                    key: "prop".into(),
995                    value_source: ValueSourceDef::Literal(Value::String("v".into())),
996                }
997            ),
998            "DeclarativeSetProperty { key: \"prop\", value_source: Literal(String(\"v\")) }"
999        );
1000
1001        assert_eq!(
1002            format!(
1003                "{:?}",
1004                BuilderStep::DeclarativeScript {
1005                    expression: expr.clone(),
1006                }
1007            ),
1008            "DeclarativeScript { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
1009        );
1010
1011        // DeclarativeFunction needs a FunctionDefinition
1012        let func_def = FunctionDefinition {
1013            id: FunctionId("test-id".into()),
1014            runtime: "my_runtime".into(),
1015            source: "${body}".into(),
1016            timeout_ms: 5000,
1017            route_id: None,
1018            step_index: None,
1019        };
1020        assert_eq!(
1021            format!(
1022                "{:?}",
1023                BuilderStep::DeclarativeFunction {
1024                    definition: func_def,
1025                }
1026            ),
1027            "DeclarativeFunction { definition: FunctionDefinition { id: FunctionId(\"test-id\"), runtime: \"my_runtime\", source: \"${body}\", timeout_ms: 5000, route_id: None, step_index: None } }"
1028        );
1029
1030        assert_eq!(
1031            format!(
1032                "{:?}",
1033                BuilderStep::WireTap {
1034                    uri: "mock:tap".into(),
1035                }
1036            ),
1037            "WireTap { uri: \"mock:tap\" }"
1038        );
1039
1040        assert_eq!(
1041            format!(
1042                "{:?}",
1043                BuilderStep::DeclarativeLog {
1044                    level: camel_processor::LogLevel::Info,
1045                    message: ValueSourceDef::Expression(expr.clone()),
1046                }
1047            ),
1048            "DeclarativeLog { level: Info, message: Expression(LanguageExpressionDef { language: \"simple\", source: \"${body}\" }) }"
1049        );
1050
1051        assert_eq!(
1052            format!(
1053                "{:?}",
1054                BuilderStep::Bean {
1055                    name: "myBean".into(),
1056                    method: "process".into(),
1057                }
1058            ),
1059            "Bean { name: \"myBean\", method: \"process\" }"
1060        );
1061
1062        assert_eq!(
1063            format!(
1064                "{:?}",
1065                BuilderStep::Script {
1066                    language: "js".into(),
1067                    script: "body".into(),
1068                }
1069            ),
1070            "Script { language: \"js\", script: \"body\" }"
1071        );
1072
1073        assert_eq!(
1074            format!(
1075                "{:?}",
1076                BuilderStep::Aggregate {
1077                    config: camel_api::AggregatorConfig::correlate_by("id")
1078                        .complete_when_size(1)
1079                        .build()
1080                        .unwrap(),
1081                }
1082            ),
1083            "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 } }"
1084        );
1085
1086        assert_eq!(
1087            format!(
1088                "{:?}",
1089                BuilderStep::DynamicRouter {
1090                    config: DynamicRouterConfig::new(Arc::new(|_: &Exchange| Some(
1091                        "mock:dr".into()
1092                    ))),
1093                }
1094            ),
1095            "DynamicRouter { config: DynamicRouterConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000, timeout: Some(60s) } }"
1096        );
1097
1098        assert_eq!(
1099            format!(
1100                "{:?}",
1101                BuilderStep::RoutingSlip {
1102                    config: RoutingSlipConfig::new(Arc::new(|_: &Exchange| Some("mock:rs".into()))),
1103                }
1104            ),
1105            "RoutingSlip { config: RoutingSlipConfig { uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false } }"
1106        );
1107
1108        assert_eq!(
1109            format!(
1110                "{:?}",
1111                BuilderStep::RecipientList {
1112                    config: RecipientListConfig::new(Arc::new(|_: &Exchange| String::new())),
1113                }
1114            ),
1115            "RecipientList { config: RecipientListConfig { delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, max_recipients: 1000 } }"
1116        );
1117
1118        assert_eq!(
1119            format!(
1120                "{:?}",
1121                BuilderStep::Enrich {
1122                    uri: "mock:enrich".into(),
1123                    strategy: Some("agg".into()),
1124                    timeout_ms: Some(1000),
1125                }
1126            ),
1127            "Enrich { uri: \"mock:enrich\", strategy: Some(\"agg\"), timeout_ms: Some(1000) }"
1128        );
1129
1130        assert_eq!(
1131            format!(
1132                "{:?}",
1133                BuilderStep::PollEnrich {
1134                    uri: "mock:poll".into(),
1135                    strategy: None,
1136                    timeout_ms: None,
1137                }
1138            ),
1139            "PollEnrich { uri: \"mock:poll\", strategy: None, timeout_ms: None }"
1140        );
1141
1142        assert_eq!(
1143            format!(
1144                "{:?}",
1145                BuilderStep::Validate {
1146                    predicate: expr.clone(),
1147                }
1148            ),
1149            "Validate { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" } }"
1150        );
1151
1152        assert_eq!(
1153            format!("{:?}", BuilderStep::Sampling { period: 100 }),
1154            "Sampling { period: 100 }"
1155        );
1156
1157        assert_eq!(
1158            format!(
1159                "{:?}",
1160                BuilderStep::Resequence {
1161                    policy_config: Default::default(),
1162                }
1163            ),
1164            "Resequence { policy_config: ResequencePolicyConfig { mode: Batch { correlation: \"header.id\", sort: \"header.id\", completion: SizeOrTimeout(100, 30000) } } }"
1165        );
1166
1167        // -- group C: variants with named fields (sub-steps is Vec) ------------
1168        assert_eq!(
1169            format!(
1170                "{:?}",
1171                BuilderStep::DeclarativeFilter {
1172                    predicate: expr.clone(),
1173                    steps: vec![BuilderStep::Stop],
1174                }
1175            ),
1176            "DeclarativeFilter { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }"
1177        );
1178
1179        assert_eq!(
1180            format!(
1181                "{:?}",
1182                BuilderStep::DeclarativeSplit {
1183                    expression: expr.clone(),
1184                    aggregation: AggregationStrategy::Original,
1185                    parallel: false,
1186                    parallel_limit: Some(2),
1187                    stop_on_exception: true,
1188                    steps: vec![BuilderStep::Stop],
1189                }
1190            ),
1191            "DeclarativeSplit { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, aggregation: Original, parallel: false, parallel_limit: Some(2), stop_on_exception: true, steps: [Stop] }"
1192        );
1193
1194        assert_eq!(
1195            format!(
1196                "{:?}",
1197                BuilderStep::Split {
1198                    config: camel_api::splitter::SplitterConfig::new(
1199                        camel_api::splitter::split_body_lines()
1200                    ),
1201                    steps: vec![BuilderStep::Stop],
1202                }
1203            ),
1204            "Split { config: SplitterConfig { expression: \"<split-expression>\", aggregation: LastWins, parallel: false, parallel_limit: None, stop_on_exception: true, max_fragments: 100000 }, steps: [Stop] }"
1205        );
1206
1207        assert_eq!(
1208            format!(
1209                "{:?}",
1210                BuilderStep::Filter {
1211                    predicate: FilterPredicate::new(|_: &Exchange| true),
1212                    steps: vec![BuilderStep::Stop],
1213                }
1214            ),
1215            "Filter { predicate: FilterPredicate(..), steps: [Stop] }"
1216        );
1217
1218        assert_eq!(
1219            format!(
1220                "{:?}",
1221                BuilderStep::Throttle {
1222                    config: camel_api::ThrottlerConfig::new(
1223                        10,
1224                        std::time::Duration::from_millis(10)
1225                    ),
1226                    steps: vec![BuilderStep::Stop],
1227                }
1228            ),
1229            "Throttle { config: ThrottlerConfig { max_requests: 10, period: 10ms, strategy: Delay }, steps: [Stop] }"
1230        );
1231
1232        assert_eq!(
1233            format!(
1234                "{:?}",
1235                BuilderStep::LoadBalance {
1236                    config: camel_api::LoadBalancerConfig::round_robin(),
1237                    steps: vec![BuilderStep::To("mock:l1".into())],
1238                }
1239            ),
1240            "LoadBalance { config: LoadBalancerConfig { strategy: RoundRobin }, steps: [To(\"mock:l1\")] }"
1241        );
1242
1243        assert_eq!(
1244            format!(
1245                "{:?}",
1246                BuilderStep::Delay {
1247                    config: camel_api::DelayConfig::new(500),
1248                }
1249            ),
1250            "Delay { config: DelayConfig { delay_ms: 500, dynamic_header: None, max_delay_ms: 3600000 } }"
1251        );
1252
1253        // -- group D: Choice / DeclarativeChoice --------------------------------
1254        // derive(Debug) emits full field enumeration; WhenStep Debug is a
1255        // full struct listing; nested BuilderStep::Stop stays as `Stop`.
1256        assert_eq!(
1257            format!(
1258                "{:?}",
1259                BuilderStep::Choice {
1260                    whens: vec![WhenStep {
1261                        predicate: FilterPredicate::new(|_: &Exchange| true),
1262                        steps: vec![BuilderStep::To("mock:a".into())],
1263                    }],
1264                    otherwise: None,
1265                }
1266            ),
1267            "Choice { whens: [WhenStep { predicate: FilterPredicate(..), steps: [To(\"mock:a\")] }], otherwise: None }"
1268        );
1269
1270        assert_eq!(
1271            format!(
1272                "{:?}",
1273                BuilderStep::DeclarativeChoice {
1274                    whens: vec![DeclarativeWhenStep {
1275                        predicate: expr.clone(),
1276                        steps: vec![BuilderStep::Stop],
1277                    }],
1278                    otherwise: Some(vec![BuilderStep::Stop]),
1279                }
1280            ),
1281            "DeclarativeChoice { whens: [DeclarativeWhenStep { predicate: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [Stop] }], otherwise: Some([Stop]) }"
1282        );
1283
1284        // -- group E: Multicast -------------------------------------------------
1285        assert_eq!(
1286            format!(
1287                "{:?}",
1288                BuilderStep::Multicast {
1289                    steps: vec![BuilderStep::To("direct:a".into())],
1290                    config: MulticastConfig::new(),
1291                }
1292            ),
1293            "Multicast { steps: [To(\"direct:a\")], config: MulticastConfig { parallel: false, parallel_limit: None, stop_on_exception: false, timeout: None, aggregation: LastWins } }"
1294        );
1295
1296        // -- group F: DeclarativeDynamicRouter / DeclarativeRoutingSlip --------
1297        assert_eq!(
1298            format!(
1299                "{:?}",
1300                BuilderStep::DeclarativeDynamicRouter {
1301                    expression: expr.clone(),
1302                    uri_delimiter: ",".into(),
1303                    cache_size: 1000,
1304                    ignore_invalid_endpoints: false,
1305                    max_iterations: 1000,
1306                }
1307            ),
1308            "DeclarativeDynamicRouter { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false, max_iterations: 1000 }"
1309        );
1310
1311        assert_eq!(
1312            format!(
1313                "{:?}",
1314                BuilderStep::DeclarativeRoutingSlip {
1315                    expression: expr.clone(),
1316                    uri_delimiter: ",".into(),
1317                    cache_size: 1000,
1318                    ignore_invalid_endpoints: false,
1319                }
1320            ),
1321            "DeclarativeRoutingSlip { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, uri_delimiter: \",\", cache_size: 1000, ignore_invalid_endpoints: false }"
1322        );
1323
1324        // -- group G: DeclarativeRecipientList ---------------------------------
1325        assert_eq!(
1326            format!(
1327                "{:?}",
1328                BuilderStep::DeclarativeRecipientList {
1329                    expression: expr.clone(),
1330                    delimiter: ",".into(),
1331                    parallel: false,
1332                    parallel_limit: None,
1333                    stop_on_exception: false,
1334                    aggregation: "original".into(),
1335                }
1336            ),
1337            "DeclarativeRecipientList { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, delimiter: \",\", parallel: false, parallel_limit: None, stop_on_exception: false, aggregation: \"original\" }"
1338        );
1339
1340        // -- group H: Loop / DeclarativeLoop -----------------------------------
1341        assert_eq!(
1342            format!(
1343                "{:?}",
1344                BuilderStep::Loop {
1345                    config: camel_api::loop_eip::LoopConfig::new(LoopMode::Count(3)),
1346                    steps: vec![],
1347                }
1348            ),
1349            "Loop { config: LoopConfig { mode: Count(3), max_iterations: 10000 }, steps: [] }"
1350        );
1351
1352        assert_eq!(
1353            format!(
1354                "{:?}",
1355                BuilderStep::DeclarativeLoop {
1356                    count: Some(5),
1357                    while_predicate: None,
1358                    steps: vec![],
1359                    max_iterations: Some(100),
1360                }
1361            ),
1362            "DeclarativeLoop { count: Some(5), while_predicate: None, steps: [], max_iterations: Some(100) }"
1363        );
1364
1365        // -- group I: ClaimCheck -----------------------------------------------
1366        assert_eq!(
1367            format!(
1368                "{:?}",
1369                BuilderStep::ClaimCheck {
1370                    repository: "myRepo".into(),
1371                    operation: "checkout".into(),
1372                    key: expr.clone(),
1373                    filter: None,
1374                }
1375            ),
1376            "ClaimCheck { repository: \"myRepo\", operation: \"checkout\", key: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, filter: None }"
1377        );
1378
1379        // -- group J: Sort -----------------------------------------------------
1380        assert_eq!(
1381            format!(
1382                "{:?}",
1383                BuilderStep::Sort {
1384                    expression: expr.clone(),
1385                    reverse: false,
1386                }
1387            ),
1388            "Sort { expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, reverse: false }"
1389        );
1390
1391        // -- group K: IdempotentConsumer ---------------------------------------
1392        assert_eq!(
1393            format!(
1394                "{:?}",
1395                BuilderStep::IdempotentConsumer {
1396                    repository: "myRepo".into(),
1397                    expression: expr.clone(),
1398                    steps: vec![],
1399                    eager: true,
1400                    remove_on_failure: false,
1401                }
1402            ),
1403            "IdempotentConsumer { repository: \"myRepo\", expression: LanguageExpressionDef { language: \"simple\", source: \"${body}\" }, steps: [], eager: true, remove_on_failure: false }"
1404        );
1405
1406        // -- group L: DeclarativeDoTry -----------------------------------------
1407        assert_eq!(
1408            format!(
1409                "{:?}",
1410                BuilderStep::DeclarativeDoTry {
1411                    try_steps: vec![BuilderStep::Stop],
1412                    catch: vec![],
1413                    finally: None,
1414                }
1415            ),
1416            "DeclarativeDoTry { try_steps: [Stop], catch: [], finally: None }"
1417        );
1418
1419        // -- group M: DeclarativeStreamSplit -----------------------------------
1420        assert_eq!(
1421            format!(
1422                "{:?}",
1423                BuilderStep::DeclarativeStreamSplit {
1424                    stream_config: StreamSplitConfig {
1425                        format: StreamSplitFormat::Ndjson,
1426                        max_record_bytes: 1024 * 1024,
1427                        batch_size: 1,
1428                        chunk_size: None,
1429                        include_origin: true,
1430                    },
1431                    aggregation: AggregationStrategy::Original,
1432                    stop_on_exception: true,
1433                    steps: vec![BuilderStep::Stop],
1434                }
1435            ),
1436            "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] }"
1437        );
1438    }
1439
1440    #[test]
1441    fn test_builder_step_multicast_variant() {
1442        use camel_api::MulticastConfig;
1443
1444        let step = BuilderStep::Multicast {
1445            steps: vec![BuilderStep::To("direct:a".into())],
1446            config: MulticastConfig::new(),
1447        };
1448
1449        assert!(matches!(step, BuilderStep::Multicast { .. }));
1450    }
1451
1452    #[test]
1453    fn test_route_definition_defaults() {
1454        let def = RouteDefinition::new("direct:test", vec![]).with_route_id("test-route");
1455        assert_eq!(def.route_id(), "test-route");
1456        assert!(def.auto_startup());
1457        assert_eq!(def.startup_order(), 1000);
1458    }
1459
1460    #[test]
1461    fn test_route_definition_builders() {
1462        let def = RouteDefinition::new("direct:test", vec![])
1463            .with_route_id("my-route")
1464            .with_auto_startup(false)
1465            .with_startup_order(50);
1466        assert_eq!(def.route_id(), "my-route");
1467        assert!(!def.auto_startup());
1468        assert_eq!(def.startup_order(), 50);
1469    }
1470
1471    #[test]
1472    fn test_route_definition_accessors_cover_core_fields() {
1473        let def = RouteDefinition::new("direct:in", vec![BuilderStep::To("mock:out".into())])
1474            .with_route_id("accessor-route");
1475
1476        assert_eq!(def.from_uri(), "direct:in");
1477        assert_eq!(def.steps().len(), 1);
1478        assert!(matches!(def.steps()[0], BuilderStep::To(_)));
1479    }
1480
1481    #[test]
1482    fn test_route_definition_error_handler_circuit_breaker_and_concurrency_accessors() {
1483        use camel_api::circuit_breaker::CircuitBreakerConfig;
1484        use camel_api::error_handler::ErrorHandlerConfig;
1485        use camel_component_api::ConcurrencyModel;
1486
1487        let def = RouteDefinition::new("direct:test", vec![])
1488            .with_route_id("eh-route")
1489            .with_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlc"))
1490            .with_circuit_breaker(CircuitBreakerConfig::new())
1491            .with_concurrency(ConcurrencyModel::Concurrent { max: Some(4) });
1492
1493        let eh = def
1494            .error_handler_config()
1495            .expect("error handler should be set");
1496        assert_eq!(eh.dlc_uri.as_deref(), Some("log:dlc"));
1497        assert!(def.circuit_breaker_config().is_some());
1498        assert!(matches!(
1499            def.concurrency_override(),
1500            Some(ConcurrencyModel::Concurrent { max: Some(4) })
1501        ));
1502    }
1503
1504    #[test]
1505    fn test_builder_step_debug_covers_many_variants() {
1506        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
1507        use camel_api::{
1508            BoxProcessor, DynamicRouterConfig, Exchange, FilterPredicate, IdentityProcessor,
1509            OpaqueProcessor, RoutingSlipConfig, Value,
1510        };
1511        use std::sync::Arc;
1512
1513        let expr = LanguageExpressionDef {
1514            language: "simple".into(),
1515            source: "${body}".into(),
1516        };
1517
1518        let steps = vec![
1519            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
1520            BuilderStep::To("mock:out".into()),
1521            BuilderStep::Stop,
1522            BuilderStep::Log {
1523                level: camel_processor::LogLevel::Info,
1524                message: "hello".into(),
1525            },
1526            BuilderStep::DeclarativeSetHeader {
1527                key: "k".into(),
1528                value: ValueSourceDef::Literal(Value::String("v".into())),
1529            },
1530            BuilderStep::DeclarativeSetBody {
1531                value: ValueSourceDef::Expression(expr.clone()),
1532            },
1533            BuilderStep::DeclarativeFilter {
1534                predicate: expr.clone(),
1535                steps: vec![BuilderStep::Stop],
1536            },
1537            BuilderStep::DeclarativeChoice {
1538                whens: vec![DeclarativeWhenStep {
1539                    predicate: expr.clone(),
1540                    steps: vec![BuilderStep::Stop],
1541                }],
1542                otherwise: Some(vec![BuilderStep::Stop]),
1543            },
1544            BuilderStep::DeclarativeScript {
1545                expression: expr.clone(),
1546            },
1547            BuilderStep::DeclarativeSplit {
1548                expression: expr.clone(),
1549                aggregation: AggregationStrategy::Original,
1550                parallel: false,
1551                parallel_limit: Some(2),
1552                stop_on_exception: true,
1553                steps: vec![BuilderStep::Stop],
1554            },
1555            BuilderStep::Split {
1556                config: SplitterConfig::new(split_body_lines()),
1557                steps: vec![BuilderStep::Stop],
1558            },
1559            BuilderStep::Aggregate {
1560                config: camel_api::AggregatorConfig::correlate_by("id")
1561                    .complete_when_size(1)
1562                    .build()
1563                    .unwrap(),
1564            },
1565            BuilderStep::Filter {
1566                predicate: FilterPredicate::new(|_: &Exchange| true),
1567                steps: vec![BuilderStep::Stop],
1568            },
1569            BuilderStep::WireTap {
1570                uri: "mock:tap".into(),
1571            },
1572            BuilderStep::DeclarativeLog {
1573                level: camel_processor::LogLevel::Info,
1574                message: ValueSourceDef::Expression(expr.clone()),
1575            },
1576            BuilderStep::Bean {
1577                name: "bean".into(),
1578                method: "call".into(),
1579            },
1580            BuilderStep::Script {
1581                language: "rhai".into(),
1582                script: "body".into(),
1583            },
1584            BuilderStep::Throttle {
1585                config: camel_api::ThrottlerConfig::new(10, std::time::Duration::from_millis(10)),
1586                steps: vec![BuilderStep::Stop],
1587            },
1588            BuilderStep::LoadBalance {
1589                config: camel_api::LoadBalancerConfig::round_robin(),
1590                steps: vec![BuilderStep::To("mock:l1".into())],
1591            },
1592            BuilderStep::DynamicRouter {
1593                config: DynamicRouterConfig::new(Arc::new(|_| Some("mock:dr".into()))),
1594            },
1595            BuilderStep::RoutingSlip {
1596                config: RoutingSlipConfig::new(Arc::new(|_| Some("mock:rs".into()))),
1597            },
1598        ];
1599
1600        for step in steps {
1601            let dbg = format!("{step:?}");
1602            assert!(!dbg.is_empty());
1603        }
1604    }
1605
1606    #[test]
1607    fn test_route_definition_to_info_preserves_metadata() {
1608        let info = RouteDefinition::new("direct:test", vec![])
1609            .with_route_id("meta-route")
1610            .with_auto_startup(false)
1611            .with_startup_order(7)
1612            .to_info();
1613
1614        assert_eq!(info.route_id(), "meta-route");
1615        assert!(!info.auto_startup());
1616        assert_eq!(info.startup_order(), 7);
1617    }
1618
1619    #[test]
1620    fn test_choice_builder_step_debug() {
1621        use camel_api::FilterPredicate;
1622
1623        fn always_true(_: &camel_api::Exchange) -> bool {
1624            true
1625        }
1626
1627        let step = BuilderStep::Choice {
1628            whens: vec![WhenStep {
1629                predicate: FilterPredicate::new(always_true),
1630                steps: vec![BuilderStep::To("mock:a".into())],
1631            }],
1632            otherwise: None,
1633        };
1634        let debug = format!("{step:?}");
1635        assert!(debug.contains("Choice"));
1636    }
1637
1638    #[test]
1639    fn test_route_definition_unit_of_work() {
1640        use camel_api::UnitOfWorkConfig;
1641        let config = UnitOfWorkConfig {
1642            on_complete: Some("log:complete".into()),
1643            on_failure: Some("log:failed".into()),
1644        };
1645        let def = RouteDefinition::new("direct:test", vec![])
1646            .with_route_id("uow-test")
1647            .with_unit_of_work(config.clone());
1648        assert_eq!(
1649            def.unit_of_work_config().unwrap().on_complete.as_deref(),
1650            Some("log:complete")
1651        );
1652        assert_eq!(
1653            def.unit_of_work_config().unwrap().on_failure.as_deref(),
1654            Some("log:failed")
1655        );
1656
1657        let def_no_uow = RouteDefinition::new("direct:test", vec![]).with_route_id("no-uow");
1658        assert!(def_no_uow.unit_of_work_config().is_none());
1659    }
1660
1661    #[test]
1662    fn test_route_definition_security_policy_accessor() {
1663        use async_trait::async_trait;
1664        use camel_api::CamelError;
1665        use camel_api::Exchange;
1666        use camel_api::security_policy::{
1667            AuthContext, AuthorizationDecision, Principal, SecurityPolicy, SecurityPolicyConfig,
1668        };
1669
1670        struct StubPolicy;
1671        #[async_trait]
1672        impl SecurityPolicy for StubPolicy {
1673            async fn evaluate(
1674                &self,
1675                _exchange: &mut Exchange,
1676                _auth: &AuthContext<'_>,
1677            ) -> Result<AuthorizationDecision, CamelError> {
1678                Ok(AuthorizationDecision::Granted {
1679                    principal: Principal {
1680                        subject: "test".into(),
1681                        issuer: "test".into(),
1682                        audience: vec![],
1683                        scopes: vec![],
1684                        roles: vec![],
1685                        claims: serde_json::Value::Null,
1686                    },
1687                })
1688            }
1689        }
1690
1691        let def_no_sp = RouteDefinition::new("direct:test", vec![]).with_route_id("no-sp");
1692        assert!(def_no_sp.security_policy_config().is_none());
1693
1694        let def = RouteDefinition::new("direct:test", vec![])
1695            .with_route_id("sp-test")
1696            .with_security_policy(SecurityPolicyConfig::new(StubPolicy));
1697        assert!(def.security_policy_config().is_some());
1698    }
1699
1700    #[test]
1701    fn test_route_definition_security_authenticator_accessor() {
1702        use camel_api::security_policy::Principal;
1703
1704        struct TestAuth;
1705        #[async_trait::async_trait]
1706        impl TokenAuthenticator for TestAuth {
1707            async fn authenticate_bearer(
1708                &self,
1709                _token: &str,
1710            ) -> Result<Principal, camel_api::CamelError> {
1711                Ok(Principal {
1712                    subject: "test".into(),
1713                    issuer: "test".into(),
1714                    audience: vec![],
1715                    scopes: vec![],
1716                    roles: vec![],
1717                    claims: serde_json::Value::Null,
1718                })
1719            }
1720        }
1721
1722        let def_no_auth = RouteDefinition::new("direct:test".to_string(), vec![]);
1723        assert!(def_no_auth.security_authenticator().is_none());
1724
1725        let auth = Arc::new(TestAuth);
1726        let def = RouteDefinition::new("direct:test".to_string(), vec![])
1727            .with_security_authenticator(auth);
1728        assert!(def.security_authenticator().is_some());
1729    }
1730
1731    #[test]
1732    fn test_map_steps_swaps_steps_and_preserves_other_fields() {
1733        let original = RouteDefinition::new(
1734            "direct:test".to_string(),
1735            vec![BuilderStep::To("mock:a".into()), BuilderStep::Stop],
1736        )
1737        .with_route_id("my-route");
1738
1739        let mapped = original.map_steps(|steps| {
1740            let mut out = Vec::with_capacity(steps.len() + 1);
1741            out.push(BuilderStep::To("mock:prefix".into()));
1742            out.extend(steps);
1743            out
1744        });
1745
1746        // Steps were transformed.
1747        assert_eq!(mapped.steps().len(), 3);
1748        assert!(matches!(mapped.steps()[0], BuilderStep::To(ref s) if s == "mock:prefix"));
1749        assert!(matches!(mapped.steps()[1], BuilderStep::To(ref s) if s == "mock:a"));
1750        // Other fields preserved.
1751        assert_eq!(mapped.route_id(), "my-route");
1752    }
1753
1754    #[test]
1755    fn circuit_breaker_fallback_accessor_returns_steps() {
1756        let def = RouteDefinition::new("direct:start", vec![])
1757            .with_circuit_breaker_fallback(vec![BuilderStep::To("mock:out".into())]);
1758        assert_eq!(def.circuit_breaker_fallback().len(), 1);
1759    }
1760}
camel_core/lifecycle/application/route_definition.rs

camel_core/lifecycle/application/
route_definition.rs
