camel_builder/
lib.rs

1//! Fluent builder API for constructing Camel routes programmatically with EIP patterns.
2//!
3//! Main types: `RouteBuilder`, `StepAccumulator`, `SplitBuilder`, `ChoiceBuilder`, `MulticastBuilder`,
4//! `ThrottleBuilder`, `LoopBuilder`, `LoadBalancerBuilder`, `OnExceptionBuilder`.
5
6use std::collections::BTreeMap;
7
8use camel_api::DelayConfig;
9use camel_api::aggregator::{
10    AggregationStrategy, AggregatorConfig, CompletionCondition, CompletionMode, CorrelationStrategy,
11};
12use camel_api::body::Body;
13use camel_api::body_converter::BodyType;
14use camel_api::circuit_breaker::CircuitBreakerConfig;
15use camel_api::dynamic_router::{DynamicRouterConfig, RouterExpression};
16use camel_api::error_handler::{ErrorHandlerConfig, RedeliveryPolicy};
17use camel_api::load_balancer::LoadBalancerConfig;
18use camel_api::loop_eip::{LoopConfig, LoopMode};
19use camel_api::multicast::{MulticastConfig, MulticastStrategy};
20use camel_api::recipient_list::{RecipientListConfig, RecipientListExpression};
21use camel_api::routing_slip::{RoutingSlipConfig, RoutingSlipExpression};
22use camel_api::splitter::SplitterConfig;
23use camel_api::throttler::{ThrottleStrategy, ThrottlerConfig};
24use camel_api::{
25    BoxProcessor, CamelError, CanonicalRouteSpec, EndpointUri, Exchange, FilterPredicate,
26    IdentityProcessor, LanguageExpressionDef, OpaqueProcessor, ProcessorFn, Value,
27    runtime::{
28        CanonicalAggregateSpec, CanonicalAggregateStrategySpec, CanonicalCircuitBreakerSpec,
29        CanonicalSplitAggregationSpec, CanonicalSplitExpressionSpec, CanonicalStepSpec,
30        CanonicalWhenSpec,
31    },
32};
33use camel_component_api::ConcurrencyModel;
34use camel_core::route::{BuilderStep, DeclarativeWhenStep, RouteDefinition, WhenStep};
35use camel_processor::{
36    ConvertBodyTo, DynamicSetHeader, LogLevel, MapBody, MarshalService, SetBody, SetHeader,
37    StreamCacheService, UnmarshalService, builtin_data_format,
38};
39
40// ── Module declarations ─────────────────────────────────────────────────────
41pub mod do_try;
42pub use do_try::{DoCatchBuilder, DoFinallyBuilder, DoTryBuilder};
43
44/// Shared step-accumulation methods for all builder types.
45///
46/// Implementors provide `steps_mut()` and get step-adding methods for free.
47/// `filter()` and other branching methods are NOT included — they return
48/// different types per builder and stay as per-builder methods.
49pub trait StepAccumulator: Sized {
50    fn steps_mut(&mut self) -> &mut Vec<BuilderStep>;
51
52    fn to(mut self, endpoint: impl Into<String>) -> Self {
53        self.steps_mut().push(BuilderStep::To(endpoint.into()));
54        self
55    }
56
57    fn process<F, Fut>(mut self, f: F) -> Self
58    where
59        F: Fn(Exchange) -> Fut + Send + Sync + 'static,
60        Fut: std::future::Future<Output = Result<Exchange, CamelError>> + Send + 'static,
61    {
62        let svc = ProcessorFn::new(f);
63        self.steps_mut()
64            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
65                svc,
66            ))));
67        self
68    }
69
70    fn process_fn(mut self, processor: BoxProcessor) -> Self {
71        self.steps_mut()
72            .push(BuilderStep::Processor(OpaqueProcessor(processor)));
73        self
74    }
75
76    fn set_header(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
77        let svc = SetHeader::new(IdentityProcessor, key, value);
78        self.steps_mut()
79            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
80                svc,
81            ))));
82        self
83    }
84
85    fn map_body<F>(mut self, mapper: F) -> Self
86    where
87        F: Fn(Body) -> Body + Clone + Send + Sync + 'static,
88    {
89        let svc = MapBody::new(IdentityProcessor, mapper);
90        self.steps_mut()
91            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
92                svc,
93            ))));
94        self
95    }
96
97    fn set_body<B>(mut self, body: B) -> Self
98    where
99        B: Into<Body> + Clone + Send + Sync + 'static,
100    {
101        let body: Body = body.into();
102        let svc = SetBody::new(IdentityProcessor, move |_ex: &Exchange| body.clone());
103        self.steps_mut()
104            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
105                svc,
106            ))));
107        self
108    }
109
110    /// Apache Camel-compatible alias for [`set_body`](Self::set_body).
111    ///
112    /// Transforms the message body using the given value. Semantically identical
113    /// to `set_body` — provided for familiarity with Apache Camel route DSLs.
114    fn transform<B>(self, body: B) -> Self
115    where
116        B: Into<Body> + Clone + Send + Sync + 'static,
117    {
118        self.set_body(body)
119    }
120
121    fn set_body_fn<F>(mut self, expr: F) -> Self
122    where
123        F: Fn(&Exchange) -> Body + Clone + Send + Sync + 'static,
124    {
125        let svc = SetBody::new(IdentityProcessor, expr);
126        self.steps_mut()
127            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
128                svc,
129            ))));
130        self
131    }
132
133    fn set_header_fn<F>(mut self, key: impl Into<String>, expr: F) -> Self
134    where
135        F: Fn(&Exchange) -> Value + Clone + Send + Sync + 'static,
136    {
137        let svc = DynamicSetHeader::new(IdentityProcessor, key, expr);
138        self.steps_mut()
139            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
140                svc,
141            ))));
142        self
143    }
144
145    fn aggregate(mut self, config: AggregatorConfig) -> Self {
146        self.steps_mut().push(BuilderStep::Aggregate { config });
147        self
148    }
149
150    /// Stop processing this exchange immediately. No further steps in the
151    /// current pipeline will run.
152    ///
153    /// Can be used at any point in the route: directly on RouteBuilder,
154    /// inside `.filter()`, inside `.split()`, etc.
155    fn stop(mut self) -> Self {
156        self.steps_mut().push(BuilderStep::Stop);
157        self
158    }
159
160    fn delay(mut self, duration: std::time::Duration) -> Self {
161        self.steps_mut().push(BuilderStep::Delay {
162            config: DelayConfig::from_duration(duration),
163        });
164        self
165    }
166
167    fn delay_with_header(
168        mut self,
169        duration: std::time::Duration,
170        header: impl Into<String>,
171    ) -> Self {
172        self.steps_mut().push(BuilderStep::Delay {
173            config: DelayConfig::from_duration_with_header(duration, header),
174        });
175        self
176    }
177
178    /// Log a message at the specified level.
179    ///
180    /// The message will be logged when an exchange passes through this step.
181    fn log(mut self, message: impl Into<String>, level: LogLevel) -> Self {
182        self.steps_mut().push(BuilderStep::Log {
183            level,
184            message: message.into(),
185        });
186        self
187    }
188
189    /// Convert the message body to the target type.
190    ///
191    /// Supported: Text ↔ Json ↔ Bytes. `Body::Stream` always fails.
192    /// Returns `TypeConversionFailed` if conversion is not possible.
193    ///
194    /// # Example
195    /// ```ignore
196    /// route.set_body(Value::String(r#"{"x":1}"#.into()))
197    ///      .convert_body_to(BodyType::Json)
198    ///      .to("direct:next")
199    /// ```
200    fn convert_body_to(mut self, target: BodyType) -> Self {
201        let svc = ConvertBodyTo::new(IdentityProcessor, target);
202        self.steps_mut()
203            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
204                svc,
205            ))));
206        self
207    }
208
209    fn stream_cache(mut self, threshold: usize) -> Self {
210        let config = camel_api::stream_cache::StreamCacheConfig::new(threshold);
211        let svc = StreamCacheService::new(IdentityProcessor, config);
212        self.steps_mut()
213            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
214                svc,
215            ))));
216        self
217    }
218
219    /// Materialize `Body::Stream` into `Body::Bytes` using the default threshold (128 KB).
220    ///
221    /// Equivalent to `.stream_cache(camel_api::stream_cache::DEFAULT_STREAM_CACHE_THRESHOLD)`.
222    fn stream_cache_default(self) -> Self {
223        self.stream_cache(camel_api::stream_cache::DEFAULT_STREAM_CACHE_THRESHOLD)
224    }
225
226    /// Marshal the message body using the specified data format.
227    ///
228    /// Supported formats: `"json"`, `"xml"`, `"csv"`, `"zip"`. Returns `Err(CamelError::Config)` if
229    /// the format name is unknown.
230    /// Converts a structured body (e.g., `Body::Json`) to a wire-format body (e.g., `Body::Text`).
231    ///
232    /// # Example
233    /// ```ignore
234    /// route.marshal("json")?.to("direct:next")
235    /// ```
236    fn marshal(mut self, format: impl Into<String>) -> Result<Self, CamelError> {
237        let name = format.into();
238        let df = builtin_data_format(&name)
239            .ok_or_else(|| CamelError::Config(format!("unknown data format: '{name}'")))?;
240        let svc = MarshalService::new(IdentityProcessor, df);
241        self.steps_mut()
242            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
243                svc,
244            ))));
245        Ok(self)
246    }
247
248    /// Unmarshal the message body using the specified data format.
249    ///
250    /// Supported formats: `"json"`, `"xml"`, `"csv"`, `"zip"`. Returns `Err(CamelError::Config)` if
251    /// the format name is unknown.
252    /// Converts a wire-format body (e.g., `Body::Text`) to a structured body (e.g., `Body::Json`).
253    ///
254    /// # Example
255    /// ```ignore
256    /// route.unmarshal("json")?.to("direct:next")
257    /// ```
258    fn unmarshal(mut self, format: impl Into<String>) -> Result<Self, CamelError> {
259        let name = format.into();
260        let df = builtin_data_format(&name)
261            .ok_or_else(|| CamelError::Config(format!("unknown data format: '{name}'")))?;
262        let svc = UnmarshalService::new(IdentityProcessor, df);
263        self.steps_mut()
264            .push(BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(
265                svc,
266            ))));
267        Ok(self)
268    }
269
270    /// Validate the exchange using a predicate expression.
271    ///
272    /// If the expression evaluates to `true`, the exchange continues.
273    /// If `false`, a `CamelError::ValidationError` is returned into the route error handler.
274    ///
275    /// # Example
276    /// ```ignore
277    /// route.validate("${body.size()} > 0").to("direct:out")
278    /// ```
279    fn validate(mut self, expression: impl Into<String>) -> Self {
280        let source = expression.into();
281        let expression = LanguageExpressionDef {
282            language: "simple".into(),
283            source,
284        };
285        self.steps_mut().push(BuilderStep::Validate {
286            predicate: expression,
287        });
288        self
289    }
290
291    /// Execute a script that can modify the exchange (headers, properties, body).
292    ///
293    /// The script has access to `headers`, `properties`, and `body` variables
294    /// and can modify them with assignment syntax: `headers["k"] = v`.
295    ///
296    /// # Example
297    /// ```ignore
298    /// // ignore: requires full CamelContext setup with registered language
299    /// route.script("rhai", r#"headers["tenant"] = "acme"; body = body + "_processed""#)
300    /// ```
301    fn script(mut self, language: impl Into<String>, script: impl Into<String>) -> Self {
302        self.steps_mut().push(BuilderStep::Script {
303            language: language.into(),
304            script: script.into(),
305        });
306        self
307    }
308
309    /// EIP-7 enrich: synchronous content enrichment via a resolved producer.
310    ///
311    /// Calls the given endpoint URI as a producer, then merges the response
312    /// back into the original exchange body using the default `UseEnrichedBody`
313    /// strategy (original headers/properties are preserved).
314    fn enrich(mut self, uri: impl Into<String>) -> Self {
315        self.steps_mut().push(BuilderStep::Enrich {
316            uri: uri.into(),
317            strategy: None,
318            timeout_ms: None,
319        });
320        self
321    }
322
323    /// EIP-7 pollEnrich: blocking poll of a PollingConsumer with timeout.
324    ///
325    /// Reads from a polling endpoint (e.g., file) and merges the result
326    /// into the exchange body using the default `UseEnrichedBody` strategy.
327    /// `timeout_ms` controls how long to wait for data.
328    fn poll_enrich(mut self, uri: impl Into<String>, timeout_ms: u64) -> Self {
329        self.steps_mut().push(BuilderStep::PollEnrich {
330            uri: uri.into(),
331            strategy: None,
332            timeout_ms: Some(timeout_ms),
333        });
334        self
335    }
336
337    fn bean(mut self, name: impl Into<String>, method: impl Into<String>) -> Self {
338        self.steps_mut().push(BuilderStep::Bean {
339            name: name.into(),
340            method: method.into(),
341        });
342        self
343    }
344}
345
346/// Identifies the endpoint slot a `.parameters()` call attaches to.
347#[derive(Debug, Clone, PartialEq, Eq)]
348enum EndpointSlot {
349    /// The route's `from` endpoint.
350    From,
351    /// A step in `RouteBuilder::steps`, by index.
352    Step(usize),
353}
354
355/// A fluent builder for constructing routes.
356///
357/// # Example
358///
359/// ```ignore
360/// let definition = RouteBuilder::from("timer:tick?period=1000")
361///     .set_header("source", Value::String("timer".into()))
362///     .filter(|ex| ex.input.body.as_text().is_some())
363///     .to("log:info?showHeaders=true")
364///     .build()?;
365/// ```
366/// `RouteBuilder` is `Clone`: a partially-built route can be cloned and reused as a
367/// template for multiple routes (mirrors Apache Camel's cloneable `RouteBuilder`).
368/// Clone is a deep copy of the step list; step closures live behind `Arc`/`BoxProcessor`
369/// so cloning shares the closure and duplicates only the light wrapper (rc-8m5o).
370#[derive(Clone)]
371pub struct RouteBuilder {
372    from_uri: String,
373    steps: Vec<BuilderStep>,
374    /// Pending `.parameters()` maps, each attached to a specific endpoint slot.
375    parameter_assignments: Vec<(EndpointSlot, BTreeMap<String, String>)>,
376    /// Misuse recorded at `.parameters()` call time and surfaced at `build()`.
377    parameter_misuse: Option<String>,
378    error_handler: Option<ErrorHandlerConfig>,
379    error_handler_mode: ErrorHandlerMode,
380    circuit_breaker_config: Option<CircuitBreakerConfig>,
381    security_policy_config: Option<camel_api::security_policy::SecurityPolicyConfig>,
382    security_authenticator: Option<std::sync::Arc<dyn camel_auth::TokenAuthenticator>>,
383    provider_registry: Option<std::sync::Arc<camel_auth::ProviderRegistry>>,
384    concurrency: Option<ConcurrencyModel>,
385    route_id: Option<String>,
386    auto_startup: Option<bool>,
387    startup_order: Option<i32>,
388}
389
390#[derive(Default, Clone)]
391enum ErrorHandlerMode {
392    #[default]
393    None,
394    ExplicitConfig,
395    Shorthand {
396        dlc_uri: Option<String>,
397        specs: Vec<OnExceptionSpec>,
398    },
399    Mixed,
400}
401
402#[derive(Clone)]
403struct OnExceptionSpec {
404    matches: std::sync::Arc<dyn Fn(&CamelError) -> bool + Send + Sync>,
405    retry: Option<RedeliveryPolicy>,
406    handled_by: Option<String>,
407}
408
409impl RouteBuilder {
410    /// Start building a route from the given source endpoint URI.
411    pub fn from(endpoint: &str) -> Self {
412        Self {
413            from_uri: endpoint.to_string(),
414            steps: Vec::new(),
415            parameter_assignments: Vec::new(),
416            parameter_misuse: None,
417            error_handler: None,
418            error_handler_mode: ErrorHandlerMode::None,
419            circuit_breaker_config: None,
420            security_policy_config: None,
421            security_authenticator: None,
422            provider_registry: None,
423            concurrency: None,
424            route_id: None,
425            auto_startup: None,
426            startup_order: None,
427        }
428    }
429
430    /// Open a filter scope. Only exchanges matching `predicate` will be processed
431    /// by the steps inside the scope. Non-matching exchanges skip the scope entirely
432    /// and continue to steps after `.end_filter()`.
433    pub fn filter<F>(self, predicate: F) -> FilterBuilder
434    where
435        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
436    {
437        FilterBuilder {
438            parent: self,
439            predicate: camel_api::FilterPredicate::new(predicate),
440            steps: vec![],
441        }
442    }
443
444    /// Open a choice scope for content-based routing.
445    ///
446    /// Within the choice, you can define multiple `.when()` clauses and an
447    /// optional `.otherwise()` clause. The first matching `when` predicate
448    /// determines which sub-pipeline executes.
449    pub fn choice(self) -> ChoiceBuilder {
450        ChoiceBuilder {
451            parent: self,
452            whens: vec![],
453            _otherwise: None,
454        }
455    }
456
457    /// Add a WireTap step that sends a clone of the exchange to the given
458    /// endpoint URI (fire-and-forget). The original exchange continues
459    /// downstream unchanged.
460    pub fn wire_tap(mut self, endpoint: &str) -> Self {
461        self.steps.push(BuilderStep::WireTap {
462            uri: endpoint.to_string(),
463        });
464        self
465    }
466
467    /// Attach a `parameters:` map to the most recent endpoint slot — the `from`
468    /// endpoint when called before any step, otherwise the last added step.
469    ///
470    /// Parameters on different endpoints each persist independently; none
471    /// overwrites another. Misuse (a second `.parameters()` on the same slot,
472    /// or a call with no pending endpoint step) is deferred and surfaced as a
473    /// `CamelError::RouteError` at `build()` / `build_canonical()`, never a panic.
474    pub fn parameters(mut self, params: BTreeMap<String, String>) -> Self {
475        let slot = if self.steps.is_empty() {
476            EndpointSlot::From
477        } else {
478            EndpointSlot::Step(self.steps.len() - 1)
479        };
480
481        if self
482            .parameter_assignments
483            .iter()
484            .any(|(existing, _)| *existing == slot)
485        {
486            self.parameter_misuse =
487                Some("multiple .parameters() calls attached to the same endpoint".to_string());
488        } else if !endpoint_slot_bears_uri(&slot, &self.steps) {
489            self.parameter_misuse =
490                Some(".parameters() called with no pending endpoint step".to_string());
491        }
492
493        self.parameter_assignments.push((slot, params));
494        self
495    }
496
497    /// Set a per-route error handler. Overrides the global error handler on `CamelContext`.
498    pub fn error_handler(mut self, config: ErrorHandlerConfig) -> Self {
499        self.error_handler_mode = match self.error_handler_mode {
500            ErrorHandlerMode::None | ErrorHandlerMode::ExplicitConfig => {
501                ErrorHandlerMode::ExplicitConfig
502            }
503            ErrorHandlerMode::Shorthand { .. } | ErrorHandlerMode::Mixed => ErrorHandlerMode::Mixed,
504        };
505        self.error_handler = Some(config);
506        self
507    }
508
509    /// Set a dead letter channel URI for shorthand error handler mode.
510    pub fn dead_letter_channel(mut self, uri: impl Into<String>) -> Self {
511        let uri = uri.into();
512        self.error_handler_mode = match self.error_handler_mode {
513            ErrorHandlerMode::None => ErrorHandlerMode::Shorthand {
514                dlc_uri: Some(uri),
515                specs: Vec::new(),
516            },
517            ErrorHandlerMode::Shorthand { specs, .. } => ErrorHandlerMode::Shorthand {
518                dlc_uri: Some(uri),
519                specs,
520            },
521            ErrorHandlerMode::ExplicitConfig | ErrorHandlerMode::Mixed => ErrorHandlerMode::Mixed,
522        };
523        self
524    }
525
526    /// Add a shorthand exception policy scope. Call `.end_on_exception()` to return to route builder.
527    pub fn on_exception<F>(mut self, matches: F) -> OnExceptionBuilder
528    where
529        F: Fn(&CamelError) -> bool + Send + Sync + 'static,
530    {
531        self.error_handler_mode = match self.error_handler_mode {
532            ErrorHandlerMode::None => ErrorHandlerMode::Shorthand {
533                dlc_uri: None,
534                specs: Vec::new(),
535            },
536            ErrorHandlerMode::ExplicitConfig | ErrorHandlerMode::Mixed => ErrorHandlerMode::Mixed,
537            shorthand @ ErrorHandlerMode::Shorthand { .. } => shorthand,
538        };
539
540        OnExceptionBuilder {
541            parent: self,
542            policy: OnExceptionSpec {
543                matches: std::sync::Arc::new(matches),
544                retry: None,
545                handled_by: None,
546            },
547        }
548    }
549
550    /// Set a circuit breaker for this route.
551    pub fn circuit_breaker(mut self, config: CircuitBreakerConfig) -> Self {
552        self.circuit_breaker_config = Some(config);
553        self
554    }
555
556    pub fn security_policy(
557        mut self,
558        config: camel_api::security_policy::SecurityPolicyConfig,
559    ) -> Self {
560        self.security_policy_config = Some(config);
561        self
562    }
563
564    pub fn security_authenticator(
565        mut self,
566        auth: std::sync::Arc<dyn camel_auth::TokenAuthenticator>,
567    ) -> Self {
568        self.security_authenticator = Some(auth);
569        self
570    }
571
572    /// Named provider registry for security plan compilation (ADR-0061):
573    /// routes declaring security resolve their provider here; staging
574    /// aborts when the registry cannot satisfy the declaration.
575    pub fn provider_registry(
576        mut self,
577        registry: std::sync::Arc<camel_auth::ProviderRegistry>,
578    ) -> Self {
579        self.provider_registry = Some(registry);
580        self
581    }
582
583    /// Override the consumer's default concurrency model.
584    ///
585    /// When set, the pipeline spawns a task per exchange, processing them
586    /// concurrently. `max` limits the number of simultaneously active
587    /// pipeline executions (0 = unbounded, channel buffer is backpressure).
588    ///
589    /// # Example
590    /// ```ignore
591    /// RouteBuilder::from("http://0.0.0.0:8080/api")
592    ///     .concurrent(16)  // max 16 in-flight pipeline executions
593    ///     .process(handle_request)
594    ///     .build()
595    /// ```
596    pub fn concurrent(mut self, max: usize) -> Self {
597        let max = if max == 0 { None } else { Some(max) };
598        self.concurrency = Some(ConcurrencyModel::Concurrent { max });
599        self
600    }
601
602    /// Force sequential processing, overriding a concurrent-capable consumer.
603    ///
604    /// Useful for HTTP routes that mutate shared state and need ordering
605    /// guarantees.
606    pub fn sequential(mut self) -> Self {
607        self.concurrency = Some(ConcurrencyModel::Sequential);
608        self
609    }
610
611    /// Set the route ID for this route.
612    ///
613    /// If not set, the route will be assigned an auto-generated ID.
614    pub fn route_id(mut self, id: impl Into<String>) -> Self {
615        self.route_id = Some(id.into());
616        self
617    }
618
619    /// Set whether this route should automatically start when the context starts.
620    ///
621    /// Default is `true`.
622    pub fn auto_startup(mut self, auto: bool) -> Self {
623        self.auto_startup = Some(auto);
624        self
625    }
626
627    /// Set the startup order for this route.
628    ///
629    /// Routes with lower values start first. Default is 1000.
630    pub fn startup_order(mut self, order: i32) -> Self {
631        self.startup_order = Some(order);
632        self
633    }
634
635    /// Begin a Splitter sub-pipeline. Steps added after this call (until
636    /// `.end_split()`) will be executed per-fragment.
637    ///
638    /// Returns a `SplitBuilder` — you cannot call `.build()` until
639    /// `.end_split()` closes the split scope (enforced by the type system).
640    pub fn split(self, config: SplitterConfig) -> SplitBuilder {
641        SplitBuilder {
642            parent: self,
643            config,
644            steps: Vec::new(),
645        }
646    }
647
648    /// Begin a Multicast sub-pipeline. Steps added after this call (until
649    /// `.end_multicast()`) will each receive a copy of the exchange.
650    ///
651    /// Returns a `MulticastBuilder` — you cannot call `.build()` until
652    /// `.end_multicast()` closes the multicast scope (enforced by the type system).
653    pub fn multicast(self) -> MulticastBuilder {
654        MulticastBuilder {
655            parent: self,
656            steps: Vec::new(),
657            config: MulticastConfig::new(),
658        }
659    }
660
661    /// Begin a Throttle sub-pipeline. Rate limits message processing to at most
662    /// `max_requests` per `period`. Steps inside the throttle scope are only
663    /// executed when the rate limit allows.
664    ///
665    /// Returns a `ThrottleBuilder` — you cannot call `.build()` until
666    /// `.end_throttle()` closes the throttle scope (enforced by the type system).
667    pub fn throttle(self, max_requests: usize, period: std::time::Duration) -> ThrottleBuilder {
668        ThrottleBuilder {
669            parent: self,
670            config: ThrottlerConfig::new(max_requests, period),
671            steps: Vec::new(),
672        }
673    }
674
675    /// Begin a Loop sub-pipeline that iterates a fixed number of times.
676    pub fn loop_count(self, count: usize) -> LoopBuilder {
677        LoopBuilder {
678            parent: self,
679            config: LoopConfig::new(LoopMode::Count(count)),
680            steps: vec![],
681        }
682    }
683
684    /// Begin a Loop sub-pipeline that iterates while a predicate is true.
685    pub fn loop_while<F>(self, predicate: F) -> LoopBuilder
686    where
687        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
688    {
689        LoopBuilder {
690            parent: self,
691            config: LoopConfig::new(LoopMode::While(camel_api::FilterPredicate::new(predicate))),
692            steps: vec![],
693        }
694    }
695
696    /// Begin a LoadBalance sub-pipeline. Distributes exchanges across multiple
697    /// endpoints using a configurable strategy (round-robin, random, weighted, failover).
698    ///
699    /// Returns a `LoadBalancerBuilder` — you cannot call `.build()` until
700    /// `.end_load_balance()` closes the load balance scope (enforced by the type system).
701    pub fn load_balance(self) -> LoadBalancerBuilder {
702        LoadBalancerBuilder {
703            parent: self,
704            config: LoadBalancerConfig::round_robin(),
705            steps: Vec::new(),
706        }
707    }
708
709    /// Add a dynamic router step that routes exchanges dynamically based on
710    /// expression evaluation at runtime.
711    ///
712    /// The expression receives the exchange and returns `Some(uri)` to route to
713    /// the next endpoint, or `None` to stop routing.
714    ///
715    /// # Example
716    /// ```ignore
717    /// RouteBuilder::from("timer:tick")
718    ///     .route_id("test-route")
719    ///     .dynamic_router(|ex| {
720    ///         ex.input.header("dest").and_then(|v| v.as_str().map(|s| s.to_string()))
721    ///     })
722    ///     .build()
723    /// ```
724    pub fn dynamic_router(self, expression: RouterExpression) -> Self {
725        self.dynamic_router_with_config(DynamicRouterConfig::new(expression))
726    }
727
728    /// Add a dynamic router step with full configuration.
729    ///
730    /// Allows customization of URI delimiter, cache size, timeout, and other options.
731    pub fn dynamic_router_with_config(mut self, config: DynamicRouterConfig) -> Self {
732        self.steps.push(BuilderStep::DynamicRouter { config });
733        self
734    }
735
736    pub fn routing_slip(self, expression: RoutingSlipExpression) -> Self {
737        self.routing_slip_with_config(RoutingSlipConfig::new(expression))
738    }
739
740    pub fn routing_slip_with_config(mut self, config: RoutingSlipConfig) -> Self {
741        self.steps.push(BuilderStep::RoutingSlip { config });
742        self
743    }
744
745    pub fn recipient_list(self, expression: RecipientListExpression) -> Self {
746        self.recipient_list_with_config(RecipientListConfig::new(expression))
747    }
748
749    pub fn recipient_list_with_config(mut self, config: RecipientListConfig) -> Self {
750        self.steps.push(BuilderStep::RecipientList { config });
751        self
752    }
753
754    /// Consume the builder and produce a [`RouteDefinition`].
755    // Duplicate route IDs are detected at `CamelContext::add_route_definition` time
756    // (RouteController rejects atomically with CamelError::RouteError).
757    pub fn build(mut self) -> Result<RouteDefinition, CamelError> {
758        validate_uri(&self.from_uri)?;
759        let route_id = self
760            .route_id
761            .filter(|s| !s.trim().is_empty())
762            .ok_or_else(|| {
763                CamelError::RouteError(
764                    "route must have a non-empty 'route_id' — call .route_id(\"name\") on the builder"
765                        .to_string(),
766                )
767            })?;
768        let resolved_error_handler = match self.error_handler_mode {
769            ErrorHandlerMode::None => self.error_handler,
770            ErrorHandlerMode::ExplicitConfig => self.error_handler,
771            ErrorHandlerMode::Mixed => {
772                return Err(CamelError::RouteError(
773                    "mixed error handler modes: cannot combine .error_handler(config) with shorthand methods".into(),
774                ));
775            }
776            ErrorHandlerMode::Shorthand { dlc_uri, specs } => {
777                let mut config = if let Some(uri) = dlc_uri {
778                    ErrorHandlerConfig::dead_letter_channel(uri)
779                } else {
780                    ErrorHandlerConfig::log_only()
781                };
782
783                for spec in specs {
784                    let matcher = spec.matches.clone();
785                    let mut builder = config.on_exception(move |e| matcher(e));
786
787                    if let Some(retry) = spec.retry {
788                        builder = builder.retry(retry.max_attempts).with_backoff(
789                            retry.initial_delay,
790                            retry.multiplier,
791                            retry.max_delay,
792                        );
793                        if retry.jitter_factor > 0.0 {
794                            builder = builder.with_jitter(retry.jitter_factor);
795                        }
796                    }
797
798                    if let Some(uri) = spec.handled_by {
799                        builder = builder.handled_by(uri);
800                    }
801
802                    config = builder.build();
803                }
804
805                Some(config)
806            }
807        };
808
809        // Deferred `.parameters()` validation and merge: misuse is a RouteError,
810        // URI merge failures surface as CamelError::EndpointUri via `?`.
811        apply_parameter_assignments(
812            &mut self.from_uri,
813            &mut self.steps,
814            std::mem::take(&mut self.parameter_assignments),
815            std::mem::take(&mut self.parameter_misuse),
816        )?;
817
818        let definition = RouteDefinition::new(self.from_uri, self.steps);
819        let definition = if let Some(eh) = resolved_error_handler {
820            definition.with_error_handler(eh)
821        } else {
822            definition
823        };
824        let definition = if let Some(cb) = self.circuit_breaker_config {
825            definition.with_circuit_breaker(cb)
826        } else {
827            definition
828        };
829        let definition = if let Some(sp) = self.security_policy_config {
830            definition.with_security_policy(sp)
831        } else {
832            definition
833        };
834        let definition = if let Some(auth) = self.security_authenticator {
835            definition.with_security_authenticator(auth)
836        } else {
837            definition
838        };
839        let definition = if let Some(registry) = self.provider_registry {
840            definition.with_provider_registry(registry)
841        } else {
842            definition
843        };
844        let definition = if let Some(concurrency) = self.concurrency {
845            definition.with_concurrency(concurrency)
846        } else {
847            definition
848        };
849        let definition = definition.with_route_id(route_id);
850        let definition = if let Some(auto) = self.auto_startup {
851            definition.with_auto_startup(auto)
852        } else {
853            definition
854        };
855        let definition = if let Some(order) = self.startup_order {
856            definition.with_startup_order(order)
857        } else {
858            definition
859        };
860        Ok(definition)
861    }
862
863    /// Compile this builder route into canonical spec.
864    pub fn build_canonical(mut self) -> Result<CanonicalRouteSpec, CamelError> {
865        validate_uri(&self.from_uri)?;
866        let route_id = self
867            .route_id
868            .filter(|s| !s.trim().is_empty())
869            .ok_or_else(|| {
870                CamelError::RouteError(
871                    "route must have a non-empty 'route_id' — call .route_id(\"name\") on the builder"
872                        .to_string(),
873                )
874            })?;
875
876        // Deferred `.parameters()` validation and merge (same path as `build()`).
877        apply_parameter_assignments(
878            &mut self.from_uri,
879            &mut self.steps,
880            std::mem::take(&mut self.parameter_assignments),
881            std::mem::take(&mut self.parameter_misuse),
882        )?;
883
884        let steps = canonicalize_steps(self.steps)?;
885        let circuit_breaker = self
886            .circuit_breaker_config
887            .map(canonicalize_circuit_breaker)
888            .transpose()?;
889
890        if self.security_policy_config.is_some() {
891            return Err(CamelError::RouteError(
892                "routes with security_policy cannot use the canonical/hot-reload path (not yet supported)"
893                    .into(),
894            ));
895        }
896
897        let spec = CanonicalRouteSpec {
898            route_id,
899            from: self.from_uri,
900            steps,
901            circuit_breaker,
902            auto_startup: None,
903            startup_order: None,
904            concurrency: None,
905            version: camel_api::CANONICAL_CONTRACT_VERSION,
906        };
907        spec.validate_contract()?;
908        Ok(spec)
909    }
910}
911
912pub struct OnExceptionBuilder {
913    parent: RouteBuilder,
914    policy: OnExceptionSpec,
915}
916
917impl OnExceptionBuilder {
918    pub fn retry(mut self, max_attempts: u32) -> Self {
919        self.policy.retry = Some(RedeliveryPolicy::new(max_attempts));
920        self
921    }
922
923    pub fn with_backoff(
924        mut self,
925        initial: std::time::Duration,
926        multiplier: f64,
927        max: std::time::Duration,
928    ) -> Self {
929        if let Some(ref mut retry) = self.policy.retry {
930            retry.initial_delay = initial;
931            retry.multiplier = multiplier;
932            retry.max_delay = max;
933        } else {
934            tracing::warn!("backoff/jitter configuration has no effect when retry_count is 0");
935        }
936        self
937    }
938
939    pub fn with_jitter(mut self, jitter_factor: f64) -> Self {
940        if let Some(ref mut retry) = self.policy.retry {
941            retry.jitter_factor = jitter_factor.clamp(0.0, 1.0);
942        } else {
943            tracing::warn!("backoff/jitter configuration has no effect when retry_count is 0");
944        }
945        self
946    }
947
948    pub fn handled_by(mut self, uri: impl Into<String>) -> Self {
949        self.policy.handled_by = Some(uri.into());
950        self
951    }
952
953    pub fn end_on_exception(mut self) -> RouteBuilder {
954        if let ErrorHandlerMode::Shorthand { ref mut specs, .. } = self.parent.error_handler_mode {
955            specs.push(self.policy);
956        }
957        self.parent
958    }
959}
960
961/// True when `slot` names an endpoint-bearing position: the `from` endpoint, or
962/// a step of one of the endpoint kinds (`To`, `WireTap`, `Enrich`, `PollEnrich`).
963fn endpoint_slot_bears_uri(slot: &EndpointSlot, steps: &[BuilderStep]) -> bool {
964    match slot {
965        EndpointSlot::From => true,
966        EndpointSlot::Step(i) => steps.get(*i).is_some_and(|step| {
967            matches!(
968                step,
969                BuilderStep::To(_)
970                    | BuilderStep::WireTap { .. }
971                    | BuilderStep::Enrich { .. }
972                    | BuilderStep::PollEnrich { .. }
973            )
974        }),
975    }
976}
977
978/// Merge each pending `.parameters()` assignment into its endpoint slot's URI.
979///
980/// A recorded misuse flag aborts first (builder misuse → `RouteError`). Each
981/// endpoint slot is then verified endpoint-bearing and its URI re-rendered via
982/// [`EndpointUri::try_from_uri_and_params`]; URI merge failures propagate as
983/// `CamelError::EndpointUri` (never folded into `RouteError`). Empty parameter
984/// maps are skipped entirely, preserving URI bytes (parity with the DSL
985/// lowering); misuse for such calls is still surfaced via `misuse`.
986fn apply_parameter_assignments(
987    from_uri: &mut String,
988    steps: &mut [BuilderStep],
989    assignments: Vec<(EndpointSlot, BTreeMap<String, String>)>,
990    misuse: Option<String>,
991) -> Result<(), CamelError> {
992    if let Some(reason) = misuse {
993        return Err(CamelError::RouteError(reason));
994    }
995
996    for (slot, params) in assignments {
997        // Empty maps are no-ops: re-rendering would still route through
998        // `EndpointUri` and could normalize URI bytes, diverging from the DSL
999        // lowering which skips empty maps for byte-identity/passthrough.
1000        if params.is_empty() {
1001            continue;
1002        }
1003
1004        let uri = match slot {
1005            EndpointSlot::From => &mut *from_uri,
1006            EndpointSlot::Step(i) => match steps.get_mut(i) {
1007                Some(BuilderStep::To(uri)) => uri,
1008                Some(BuilderStep::WireTap { uri, .. }) => uri,
1009                Some(BuilderStep::Enrich { uri, .. }) => uri,
1010                Some(BuilderStep::PollEnrich { uri, .. }) => uri,
1011                Some(other) => {
1012                    return Err(CamelError::RouteError(format!(
1013                        ".parameters() attached to a non-endpoint step (`{}`)",
1014                        canonical_step_name(other)
1015                    )));
1016                }
1017                None => {
1018                    return Err(CamelError::RouteError(
1019                        ".parameters() attached to an out-of-range step index".to_string(),
1020                    ));
1021                }
1022            },
1023        };
1024        let merged = EndpointUri::try_from_uri_and_params(uri, params)?.to_canonical_string();
1025        *uri = merged;
1026    }
1027
1028    Ok(())
1029}
1030
1031/// Validate that a URI is non-empty and contains a scheme component.
1032fn validate_uri(uri: &str) -> Result<(), CamelError> {
1033    let trimmed = uri.trim();
1034    if trimmed.is_empty() {
1035        return Err(CamelError::RouteError(
1036            "route must have a 'from' URI".to_string(),
1037        ));
1038    }
1039    if !trimmed.contains(':') {
1040        return Err(CamelError::RouteError(
1041            "URI must have a scheme (e.g. 'timer:tick')".to_string(),
1042        ));
1043    }
1044    let scheme = trimmed.split(':').next().unwrap_or("");
1045    if scheme.trim().is_empty() {
1046        return Err(CamelError::RouteError(
1047            "URI scheme must not be empty".to_string(),
1048        ));
1049    }
1050    Ok(())
1051}
1052
1053fn canonicalize_steps(steps: Vec<BuilderStep>) -> Result<Vec<CanonicalStepSpec>, CamelError> {
1054    let mut canonical = Vec::with_capacity(steps.len());
1055    for step in steps {
1056        canonical.push(canonicalize_step(step)?);
1057    }
1058    Ok(canonical)
1059}
1060
1061fn canonicalize_step(step: BuilderStep) -> Result<CanonicalStepSpec, CamelError> {
1062    match step {
1063        BuilderStep::To(uri) => Ok(CanonicalStepSpec::To { uri }),
1064        BuilderStep::Log { message, .. } => Ok(CanonicalStepSpec::Log { message }),
1065        BuilderStep::Stop => Ok(CanonicalStepSpec::Stop),
1066        BuilderStep::WireTap { uri } => Ok(CanonicalStepSpec::WireTap { uri }),
1067        BuilderStep::Delay { config } => Ok(CanonicalStepSpec::Delay {
1068            delay_ms: config.delay_ms,
1069            dynamic_header: config.dynamic_header,
1070        }),
1071        BuilderStep::DeclarativeScript { expression } => {
1072            Ok(CanonicalStepSpec::Script { expression })
1073        }
1074        BuilderStep::DeclarativeFilter { predicate, steps } => Ok(CanonicalStepSpec::Filter {
1075            predicate,
1076            steps: canonicalize_steps(steps)?,
1077        }),
1078        BuilderStep::DeclarativeChoice { whens, otherwise } => {
1079            let mut canonical_whens = Vec::with_capacity(whens.len());
1080            for DeclarativeWhenStep { predicate, steps } in whens {
1081                canonical_whens.push(CanonicalWhenSpec {
1082                    predicate,
1083                    steps: canonicalize_steps(steps)?,
1084                });
1085            }
1086            let otherwise = match otherwise {
1087                Some(steps) => Some(canonicalize_steps(steps)?),
1088                None => None,
1089            };
1090            Ok(CanonicalStepSpec::Choice {
1091                whens: canonical_whens,
1092                otherwise,
1093            })
1094        }
1095        BuilderStep::DeclarativeSplit {
1096            expression,
1097            aggregation,
1098            parallel,
1099            parallel_limit,
1100            stop_on_exception,
1101            steps,
1102        } => Ok(CanonicalStepSpec::Split {
1103            expression: CanonicalSplitExpressionSpec::Language(expression),
1104            aggregation: canonicalize_split_aggregation(aggregation)?,
1105            parallel,
1106            parallel_limit,
1107            stop_on_exception,
1108            steps: canonicalize_steps(steps)?,
1109        }),
1110        BuilderStep::Aggregate { config } => Ok(CanonicalStepSpec::Aggregate(
1111            canonicalize_aggregate(config)?,
1112        )),
1113        other => {
1114            let step_name = canonical_step_name(&other);
1115            let detail = camel_api::canonical_contract_rejection_reason(step_name)
1116                .unwrap_or("not included in canonical v2");
1117            Err(CamelError::RouteError(format!(
1118                "canonical v2 does not support step `{step_name}`: {detail}"
1119            )))
1120        }
1121    }
1122}
1123
1124fn canonicalize_split_aggregation(
1125    strategy: camel_api::splitter::AggregationStrategy,
1126) -> Result<CanonicalSplitAggregationSpec, CamelError> {
1127    match strategy {
1128        camel_api::splitter::AggregationStrategy::LastWins => {
1129            Ok(CanonicalSplitAggregationSpec::LastWins)
1130        }
1131        camel_api::splitter::AggregationStrategy::CollectAll => {
1132            Ok(CanonicalSplitAggregationSpec::CollectAll)
1133        }
1134        camel_api::splitter::AggregationStrategy::Custom(_) => Err(CamelError::RouteError(
1135            "canonical v2 does not support custom split aggregation".to_string(),
1136        )),
1137        camel_api::splitter::AggregationStrategy::Original => {
1138            Ok(CanonicalSplitAggregationSpec::Original)
1139        }
1140        _ => Err(CamelError::RouteError(
1141            "canonical v2 does not support this split aggregation strategy".to_string(),
1142        )),
1143    }
1144}
1145
1146fn extract_completion_fields(
1147    mode: &CompletionMode,
1148) -> Result<(Option<usize>, Option<u64>), CamelError> {
1149    match mode {
1150        CompletionMode::Single(cond) => match cond {
1151            CompletionCondition::Size(n) => Ok((Some(*n), None)),
1152            CompletionCondition::Timeout(d) => Ok((None, Some(d.as_millis() as u64))),
1153            CompletionCondition::Predicate(_) | CompletionCondition::PredicateExpr { .. } => {
1154                Err(CamelError::RouteError(
1155                    "aggregate PredicateExpr/Predicate completion cannot reverse-map to canonical \
1156                     (forward-only in rc-zit); build the canonical spec directly"
1157                        .to_string(),
1158                ))
1159            }
1160            _ => Err(CamelError::RouteError(
1161                "unsupported completion condition".to_string(),
1162            )),
1163        },
1164        CompletionMode::Any(conds) => {
1165            let mut size = None;
1166            let mut timeout_ms = None;
1167            for cond in conds {
1168                match cond {
1169                    CompletionCondition::Size(n) => size = Some(*n),
1170                    CompletionCondition::Timeout(d) => timeout_ms = Some(d.as_millis() as u64),
1171                    CompletionCondition::Predicate(_)
1172                    | CompletionCondition::PredicateExpr { .. } => {
1173                        return Err(CamelError::RouteError(
1174                            "aggregate PredicateExpr/Predicate completion cannot reverse-map to \
1175                             canonical (forward-only in rc-zit); build the canonical spec directly"
1176                                .to_string(),
1177                        ));
1178                    }
1179                    _ => {
1180                        return Err(CamelError::RouteError(
1181                            "unsupported completion condition".to_string(),
1182                        ));
1183                    }
1184                }
1185            }
1186            Ok((size, timeout_ms))
1187        }
1188        _ => Err(CamelError::RouteError(
1189            "unsupported completion mode".to_string(),
1190        )),
1191    }
1192}
1193
1194fn canonicalize_aggregate(config: AggregatorConfig) -> Result<CanonicalAggregateSpec, CamelError> {
1195    let (completion_size, completion_timeout_ms) = extract_completion_fields(&config.completion)?;
1196
1197    let header = match &config.correlation {
1198        CorrelationStrategy::HeaderName(h) => h.clone(),
1199        CorrelationStrategy::Expression { expr, .. } => expr.clone(),
1200        CorrelationStrategy::Fn(_) => {
1201            return Err(CamelError::RouteError(
1202                "canonical v2 does not support Fn correlation strategy".to_string(),
1203            ));
1204        }
1205        _ => {
1206            return Err(CamelError::RouteError(
1207                "canonical v2 does not support this correlation strategy".to_string(),
1208            ));
1209        }
1210    };
1211
1212    let correlation_key = match &config.correlation {
1213        CorrelationStrategy::HeaderName(_) => None,
1214        CorrelationStrategy::Expression { expr, .. } => Some(expr.clone()),
1215        // INVARIANT: every other correlation variant (Fn and any future
1216        // variant) is rejected by the `header` match above, which returns
1217        // early — so this branch is unreachable here.
1218        _ => unreachable!(),
1219    };
1220
1221    let strategy = match config.strategy {
1222        AggregationStrategy::CollectAll => CanonicalAggregateStrategySpec::CollectAll,
1223        AggregationStrategy::Custom(_) => {
1224            return Err(CamelError::RouteError(
1225                "canonical v2 does not support custom aggregate strategy".to_string(),
1226            ));
1227        }
1228        _ => {
1229            return Err(CamelError::RouteError(
1230                "canonical v2 does not support this aggregate strategy".to_string(),
1231            ));
1232        }
1233    };
1234    let bucket_ttl_ms = config
1235        .bucket_ttl
1236        .map(|ttl| u64::try_from(ttl.as_millis()).unwrap_or(u64::MAX));
1237
1238    Ok(CanonicalAggregateSpec {
1239        header,
1240        completion_size,
1241        completion_timeout_ms,
1242        correlation_key,
1243        force_completion_on_stop: if config.force_completion_on_stop {
1244            Some(true)
1245        } else {
1246            None
1247        },
1248        discard_on_timeout: if config.discard_on_timeout {
1249            Some(true)
1250        } else {
1251            None
1252        },
1253        strategy,
1254        max_buckets: config.max_buckets,
1255        max_bucket_size: config.max_bucket_size,
1256        bucket_ttl_ms,
1257        completion_predicate: None,
1258    })
1259}
1260
1261fn canonicalize_circuit_breaker(
1262    config: CircuitBreakerConfig,
1263) -> Result<CanonicalCircuitBreakerSpec, CamelError> {
1264    if config.fallback.is_some() {
1265        return Err(CamelError::RouteError(
1266            "canonical v2 does not support circuit breaker `fallback` (opaque BoxProcessor \
1267             cannot reverse-map to canonical steps); build the canonical spec directly"
1268                .to_string(),
1269        ));
1270    }
1271    Ok(CanonicalCircuitBreakerSpec {
1272        failure_threshold: config.failure_threshold,
1273        open_duration_ms: u64::try_from(config.open_duration.as_millis()).unwrap_or(u64::MAX),
1274        fallback: Vec::new(),
1275    })
1276}
1277
1278fn canonical_step_name(step: &BuilderStep) -> &'static str {
1279    match step {
1280        BuilderStep::Processor(_) => "processor",
1281        BuilderStep::To(_) => "to",
1282        BuilderStep::Stop => "stop",
1283        BuilderStep::Log { .. } => "log",
1284        BuilderStep::DeclarativeSetHeader { .. } => "set_header",
1285        BuilderStep::DeclarativeSetHeaderIfAbsent { .. } => "set_header_if_absent",
1286        BuilderStep::DeclarativeRemoveHeader { .. } => "remove_header",
1287        BuilderStep::DeclarativeSetBody { .. } => "set_body",
1288        BuilderStep::DeclarativeFilter { .. } => "filter",
1289        BuilderStep::DeclarativeChoice { .. } => "choice",
1290        BuilderStep::DeclarativeScript { .. } => "script",
1291        BuilderStep::DeclarativeFunction { .. } => "function",
1292        BuilderStep::DeclarativeSplit { .. } => "split",
1293        BuilderStep::Split { .. } => "split",
1294        BuilderStep::Loop { .. } | BuilderStep::DeclarativeLoop { .. } => "loop",
1295        BuilderStep::Aggregate { .. } => "aggregate",
1296        BuilderStep::Filter { .. } => "filter",
1297        BuilderStep::Choice { .. } => "choice",
1298        BuilderStep::WireTap { .. } => "wire_tap",
1299        BuilderStep::Delay { .. } => "delay",
1300        BuilderStep::Multicast { .. } => "multicast",
1301        BuilderStep::DeclarativeLog { .. } => "log",
1302        BuilderStep::Bean { .. } => "bean",
1303        BuilderStep::Script { .. } => "script",
1304        BuilderStep::Throttle { .. } => "throttle",
1305        BuilderStep::LoadBalance { .. } => "load_balancer",
1306        BuilderStep::DynamicRouter { .. } => "dynamic_router",
1307        BuilderStep::RoutingSlip { .. } => "routing_slip",
1308        BuilderStep::DeclarativeDynamicRouter { .. } => "declarative_dynamic_router",
1309        BuilderStep::DeclarativeRoutingSlip { .. } => "declarative_routing_slip",
1310        BuilderStep::RecipientList { .. } => "recipient_list",
1311        BuilderStep::DeclarativeRecipientList { .. } => "declarative_recipient_list",
1312        BuilderStep::DeclarativeSetProperty { .. } => "set_property",
1313        BuilderStep::DeclarativeStreamSplit { .. } => "stream_split",
1314        BuilderStep::Enrich { .. } => "enrich",
1315        BuilderStep::PollEnrich { .. } => "poll_enrich",
1316        BuilderStep::Validate { .. } => "validate",
1317        BuilderStep::IdempotentConsumer { .. } => "idempotent_consumer",
1318        BuilderStep::ClaimCheck { .. } => "claim_check",
1319        BuilderStep::Cache { .. } => "cache",
1320        BuilderStep::CacheInvalidate { .. } => "cache_invalidate",
1321        BuilderStep::CacheClear { .. } => "cache_clear",
1322        BuilderStep::CacheStats { .. } => "cache_stats",
1323        BuilderStep::CachePeekStale { .. } => "cache_peek_stale",
1324        BuilderStep::Sampling { .. } => "sampling",
1325        BuilderStep::Sort { .. } => "sort",
1326        BuilderStep::DeclarativeDoTry { .. } => "do_try",
1327        BuilderStep::Resequence { .. } => "resequence",
1328    }
1329}
1330
1331impl StepAccumulator for RouteBuilder {
1332    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1333        &mut self.steps
1334    }
1335}
1336
1337/// Builder for the sub-pipeline within a `.split()` ... `.end_split()` block.
1338///
1339/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1340/// but NOT `.build()` and NOT `.split()` (no nested splits).
1341///
1342/// Calling `.end_split()` packages the sub-steps into a `BuilderStep::Split`
1343/// and returns the parent `RouteBuilder`.
1344pub struct SplitBuilder {
1345    parent: RouteBuilder,
1346    config: SplitterConfig,
1347    steps: Vec<BuilderStep>,
1348}
1349
1350impl SplitBuilder {
1351    /// Open a filter scope within the split sub-pipeline.
1352    pub fn filter<F>(self, predicate: F) -> FilterInSplitBuilder
1353    where
1354        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
1355    {
1356        FilterInSplitBuilder {
1357            parent: self,
1358            predicate: camel_api::FilterPredicate::new(predicate),
1359            steps: vec![],
1360        }
1361    }
1362
1363    /// Close the split scope. Packages the accumulated sub-steps into a
1364    /// `BuilderStep::Split` and returns the parent `RouteBuilder`.
1365    pub fn end_split(mut self) -> RouteBuilder {
1366        let split_step = BuilderStep::Split {
1367            config: self.config,
1368            steps: self.steps,
1369        };
1370        self.parent.steps.push(split_step);
1371        self.parent
1372    }
1373}
1374
1375impl StepAccumulator for SplitBuilder {
1376    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1377        &mut self.steps
1378    }
1379}
1380
1381/// Builder for the sub-pipeline within a `.filter()` ... `.end_filter()` block.
1382pub struct FilterBuilder {
1383    parent: RouteBuilder,
1384    predicate: FilterPredicate,
1385    steps: Vec<BuilderStep>,
1386}
1387
1388impl FilterBuilder {
1389    /// Close the filter scope. Packages the accumulated sub-steps into a
1390    /// `BuilderStep::Filter` and returns the parent `RouteBuilder`.
1391    pub fn end_filter(mut self) -> RouteBuilder {
1392        let step = BuilderStep::Filter {
1393            predicate: self.predicate,
1394            steps: self.steps,
1395        };
1396        self.parent.steps.push(step);
1397        self.parent
1398    }
1399}
1400
1401impl StepAccumulator for FilterBuilder {
1402    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1403        &mut self.steps
1404    }
1405}
1406
1407/// Builder for a filter scope nested inside a `.split()` block.
1408pub struct FilterInSplitBuilder {
1409    parent: SplitBuilder,
1410    predicate: FilterPredicate,
1411    steps: Vec<BuilderStep>,
1412}
1413
1414impl FilterInSplitBuilder {
1415    /// Close the filter scope and return the parent `SplitBuilder`.
1416    pub fn end_filter(mut self) -> SplitBuilder {
1417        let step = BuilderStep::Filter {
1418            predicate: self.predicate,
1419            steps: self.steps,
1420        };
1421        self.parent.steps.push(step);
1422        self.parent
1423    }
1424}
1425
1426impl StepAccumulator for FilterInSplitBuilder {
1427    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1428        &mut self.steps
1429    }
1430}
1431
1432// ── Choice/When/Otherwise builders ─────────────────────────────────────────
1433
1434/// Builder for a `.choice()` ... `.end_choice()` block.
1435///
1436/// Accumulates `when` clauses and an optional `otherwise` clause.
1437/// Cannot call `.build()` until `.end_choice()` is called.
1438pub struct ChoiceBuilder {
1439    parent: RouteBuilder,
1440    whens: Vec<WhenStep>,
1441    _otherwise: Option<Vec<BuilderStep>>,
1442}
1443
1444impl ChoiceBuilder {
1445    /// Open a `when` clause. Only exchanges matching `predicate` will be
1446    /// processed by the steps inside the `.when()` ... `.end_when()` scope.
1447    pub fn when<F>(self, predicate: F) -> WhenBuilder
1448    where
1449        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
1450    {
1451        WhenBuilder {
1452            parent: self,
1453            predicate: camel_api::FilterPredicate::new(predicate),
1454            steps: vec![],
1455        }
1456    }
1457
1458    /// Open an `otherwise` clause. Executed when no `when` predicate matched.
1459    ///
1460    /// Only one `otherwise` is allowed per `choice`. Call this after all `.when()` clauses.
1461    pub fn otherwise(self) -> OtherwiseBuilder {
1462        OtherwiseBuilder {
1463            parent: self,
1464            steps: vec![],
1465        }
1466    }
1467
1468    /// Close the choice scope. Packages all accumulated `when` clauses and
1469    /// optional `otherwise` into a `BuilderStep::Choice` and returns the
1470    /// parent `RouteBuilder`.
1471    pub fn end_choice(mut self) -> RouteBuilder {
1472        let step = BuilderStep::Choice {
1473            whens: self.whens,
1474            otherwise: self._otherwise,
1475        };
1476        self.parent.steps.push(step);
1477        self.parent
1478    }
1479}
1480
1481/// Builder for the sub-pipeline within a `.when()` ... `.end_when()` block.
1482pub struct WhenBuilder {
1483    parent: ChoiceBuilder,
1484    predicate: camel_api::FilterPredicate,
1485    steps: Vec<BuilderStep>,
1486}
1487
1488impl WhenBuilder {
1489    /// Close the when scope. Packages the accumulated sub-steps into a
1490    /// `WhenStep` and returns the parent `ChoiceBuilder`.
1491    pub fn end_when(mut self) -> ChoiceBuilder {
1492        self.parent.whens.push(WhenStep {
1493            predicate: self.predicate,
1494            steps: self.steps,
1495        });
1496        self.parent
1497    }
1498}
1499
1500impl StepAccumulator for WhenBuilder {
1501    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1502        &mut self.steps
1503    }
1504}
1505
1506/// Builder for the sub-pipeline within an `.otherwise()` ... `.end_otherwise()` block.
1507pub struct OtherwiseBuilder {
1508    parent: ChoiceBuilder,
1509    steps: Vec<BuilderStep>,
1510}
1511
1512impl OtherwiseBuilder {
1513    /// Close the otherwise scope and return the parent `ChoiceBuilder`.
1514    pub fn end_otherwise(self) -> ChoiceBuilder {
1515        let OtherwiseBuilder { mut parent, steps } = self;
1516        parent._otherwise = Some(steps);
1517        parent
1518    }
1519}
1520
1521impl StepAccumulator for OtherwiseBuilder {
1522    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1523        &mut self.steps
1524    }
1525}
1526
1527/// Builder for the sub-pipeline within a `.multicast()` ... `.end_multicast()` block.
1528///
1529/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1530/// but NOT `.build()` and NOT `.multicast()` (no nested multicasts).
1531///
1532/// Calling `.end_multicast()` packages the sub-steps into a `BuilderStep::Multicast`
1533/// and returns the parent `RouteBuilder`.
1534pub struct MulticastBuilder {
1535    parent: RouteBuilder,
1536    steps: Vec<BuilderStep>,
1537    config: MulticastConfig,
1538}
1539
1540impl MulticastBuilder {
1541    pub fn parallel(mut self, parallel: bool) -> Self {
1542        self.config = self.config.parallel(parallel);
1543        self
1544    }
1545
1546    pub fn parallel_limit(mut self, limit: usize) -> Self {
1547        self.config = self.config.parallel_limit(limit);
1548        self
1549    }
1550
1551    pub fn stop_on_exception(mut self, stop: bool) -> Self {
1552        self.config = self.config.stop_on_exception(stop);
1553        self
1554    }
1555
1556    pub fn timeout(mut self, duration: std::time::Duration) -> Self {
1557        self.config = self.config.timeout(duration);
1558        self
1559    }
1560
1561    pub fn aggregation(mut self, strategy: MulticastStrategy) -> Self {
1562        self.config = self.config.aggregation(strategy);
1563        self
1564    }
1565
1566    pub fn end_multicast(mut self) -> RouteBuilder {
1567        let step = BuilderStep::Multicast {
1568            steps: self.steps,
1569            config: self.config,
1570        };
1571        self.parent.steps.push(step);
1572        self.parent
1573    }
1574}
1575
1576impl StepAccumulator for MulticastBuilder {
1577    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1578        &mut self.steps
1579    }
1580}
1581
1582/// Builder for the sub-pipeline within a `.throttle()` ... `.end_throttle()` block.
1583///
1584/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1585/// but NOT `.build()` and NOT `.throttle()` (no nested throttles).
1586///
1587/// Calling `.end_throttle()` packages the sub-steps into a `BuilderStep::Throttle`
1588/// and returns the parent `RouteBuilder`.
1589pub struct ThrottleBuilder {
1590    parent: RouteBuilder,
1591    config: ThrottlerConfig,
1592    steps: Vec<BuilderStep>,
1593}
1594
1595impl ThrottleBuilder {
1596    /// Set the throttle strategy. Default is `Delay`.
1597    ///
1598    /// - `Delay`: Queue messages until capacity available
1599    /// - `Reject`: Return error immediately when throttled
1600    /// - `Drop`: Silently discard excess messages
1601    pub fn strategy(mut self, strategy: ThrottleStrategy) -> Self {
1602        self.config = self.config.strategy(strategy);
1603        self
1604    }
1605
1606    /// Close the throttle scope. Packages the accumulated sub-steps into a
1607    /// `BuilderStep::Throttle` and returns the parent `RouteBuilder`.
1608    pub fn end_throttle(mut self) -> RouteBuilder {
1609        let step = BuilderStep::Throttle {
1610            config: self.config,
1611            steps: self.steps,
1612        };
1613        self.parent.steps.push(step);
1614        self.parent
1615    }
1616}
1617
1618impl StepAccumulator for ThrottleBuilder {
1619    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1620        &mut self.steps
1621    }
1622}
1623
1624/// Builder for the sub-pipeline within a `.loop_count()` / `.loop_while()` ... `.end_loop()` block.
1625pub struct LoopBuilder {
1626    parent: RouteBuilder,
1627    config: LoopConfig,
1628    steps: Vec<BuilderStep>,
1629}
1630
1631impl LoopBuilder {
1632    pub fn loop_count(self, count: usize) -> LoopInLoopBuilder {
1633        LoopInLoopBuilder {
1634            parent: self,
1635            config: LoopConfig::new(LoopMode::Count(count)),
1636            steps: vec![],
1637        }
1638    }
1639
1640    pub fn loop_while<F>(self, predicate: F) -> LoopInLoopBuilder
1641    where
1642        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
1643    {
1644        LoopInLoopBuilder {
1645            parent: self,
1646            config: LoopConfig::new(LoopMode::While(camel_api::FilterPredicate::new(predicate))),
1647            steps: vec![],
1648        }
1649    }
1650
1651    pub fn end_loop(mut self) -> RouteBuilder {
1652        let step = BuilderStep::Loop {
1653            config: self.config,
1654            steps: self.steps,
1655        };
1656        self.parent.steps.push(step);
1657        self.parent
1658    }
1659}
1660
1661impl StepAccumulator for LoopBuilder {
1662    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1663        &mut self.steps
1664    }
1665}
1666
1667pub struct LoopInLoopBuilder {
1668    parent: LoopBuilder,
1669    config: LoopConfig,
1670    steps: Vec<BuilderStep>,
1671}
1672
1673impl LoopInLoopBuilder {
1674    pub fn end_loop(mut self) -> LoopBuilder {
1675        let step = BuilderStep::Loop {
1676            config: self.config,
1677            steps: self.steps,
1678        };
1679        self.parent.steps.push(step);
1680        self.parent
1681    }
1682}
1683
1684impl StepAccumulator for LoopInLoopBuilder {
1685    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1686        &mut self.steps
1687    }
1688}
1689
1690/// Builder for the sub-pipeline within a `.load_balance()` ... `.end_load_balance()` block.
1691///
1692/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1693/// but NOT `.build()` and NOT `.load_balance()` (no nested load balancers).
1694///
1695/// Calling `.end_load_balance()` packages the sub-steps into a `BuilderStep::LoadBalance`
1696/// and returns the parent `RouteBuilder`.
1697pub struct LoadBalancerBuilder {
1698    parent: RouteBuilder,
1699    config: LoadBalancerConfig,
1700    steps: Vec<BuilderStep>,
1701}
1702
1703impl LoadBalancerBuilder {
1704    /// Set the load balance strategy to round-robin (default).
1705    pub fn round_robin(mut self) -> Self {
1706        self.config = LoadBalancerConfig::round_robin();
1707        self
1708    }
1709
1710    /// Set the load balance strategy to random selection.
1711    pub fn random(mut self) -> Self {
1712        self.config = LoadBalancerConfig::random();
1713        self
1714    }
1715
1716    /// Set the load balance strategy to weighted selection.
1717    ///
1718    /// Each endpoint is assigned a weight that determines its probability
1719    /// of being selected.
1720    pub fn weighted(mut self, weights: Vec<(String, u32)>) -> Self {
1721        self.config = LoadBalancerConfig::weighted(weights);
1722        self
1723    }
1724
1725    /// Set the load balance strategy to failover.
1726    ///
1727    /// Exchanges are sent to the first endpoint; on failure, the next endpoint
1728    /// is tried.
1729    pub fn failover(mut self) -> Self {
1730        self.config = LoadBalancerConfig::failover();
1731        self
1732    }
1733
1734    /// Close the load balance scope. Packages the accumulated sub-steps into a
1735    /// `BuilderStep::LoadBalance` and returns the parent `RouteBuilder`.
1736    pub fn end_load_balance(mut self) -> RouteBuilder {
1737        let step = BuilderStep::LoadBalance {
1738            config: self.config,
1739            steps: self.steps,
1740        };
1741        self.parent.steps.push(step);
1742        self.parent
1743    }
1744}
1745
1746impl StepAccumulator for LoadBalancerBuilder {
1747    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1748        &mut self.steps
1749    }
1750}
1751
1752// ---------------------------------------------------------------------------
1753// Tests
1754// ---------------------------------------------------------------------------
1755
1756#[cfg(test)]
1757mod tests {
1758    use super::*;
1759    use camel_api::SpanKindHint;
1760    use camel_api::error_handler::ErrorHandlerConfig;
1761    use camel_api::load_balancer::LoadBalanceStrategy;
1762    use camel_api::{Exchange, Message};
1763    use camel_core::route::BuilderStep;
1764    use std::sync::Arc;
1765    use std::time::Duration;
1766    use tower::{Service, ServiceExt};
1767
1768    #[test]
1769    fn test_builder_from_creates_definition() {
1770        let definition = RouteBuilder::from("timer:tick")
1771            .route_id("test-route")
1772            .build()
1773            .unwrap();
1774        assert_eq!(definition.from_uri(), "timer:tick");
1775    }
1776
1777    #[test]
1778    fn test_builder_empty_from_uri_errors() {
1779        let result = RouteBuilder::from("").route_id("test-route").build();
1780        assert!(result.is_err());
1781    }
1782
1783    #[test]
1784    fn test_build_rejects_schemeless_uri() {
1785        let result = RouteBuilder::from("no-scheme-here")
1786            .route_id("test-route")
1787            .build();
1788        match result {
1789            Err(err) => {
1790                let err_msg = format!("{err}");
1791                assert!(
1792                    err_msg.contains("scheme"),
1793                    "expected scheme-related error, got: {err_msg}"
1794                );
1795            }
1796            Ok(_) => panic!("schemeless URI should fail"),
1797        }
1798    }
1799
1800    #[test]
1801    fn test_build_rejects_empty_scheme_uri() {
1802        let result = RouteBuilder::from(":missing-scheme")
1803            .route_id("test-route")
1804            .build();
1805        match result {
1806            Err(err) => {
1807                let err_msg = format!("{err}");
1808                assert!(
1809                    err_msg.contains("scheme"),
1810                    "expected scheme-related error, got: {err_msg}"
1811                );
1812            }
1813            Ok(_) => panic!("empty-scheme URI should fail"),
1814        }
1815    }
1816
1817    #[test]
1818    fn test_build_accepts_valid_uri() {
1819        let result = RouteBuilder::from("timer:tick")
1820            .route_id("test-route")
1821            .build();
1822        assert!(result.is_ok());
1823    }
1824
1825    #[test]
1826    fn test_build_canonical_rejects_schemeless_uri() {
1827        let result = RouteBuilder::from("no-scheme-here")
1828            .route_id("test-route")
1829            .build_canonical();
1830        assert!(result.is_err());
1831    }
1832
1833    #[test]
1834    fn test_builder_to_adds_step() {
1835        let definition = RouteBuilder::from("timer:tick")
1836            .route_id("test-route")
1837            .to("log:info")
1838            .build()
1839            .unwrap();
1840
1841        assert_eq!(definition.from_uri(), "timer:tick");
1842        // We can verify steps were added by checking the structure
1843        assert!(matches!(&definition.steps()[0], BuilderStep::To(uri) if uri == "log:info"));
1844    }
1845
1846    #[test]
1847    fn test_builder_filter_adds_filter_step() {
1848        let definition = RouteBuilder::from("timer:tick")
1849            .route_id("test-route")
1850            .filter(|_ex| true)
1851            .to("mock:result")
1852            .end_filter()
1853            .build()
1854            .unwrap();
1855
1856        assert!(matches!(&definition.steps()[0], BuilderStep::Filter { .. }));
1857    }
1858
1859    #[test]
1860    fn test_builder_set_header_adds_processor_step() {
1861        let definition = RouteBuilder::from("timer:tick")
1862            .route_id("test-route")
1863            .set_header("key", Value::String("value".into()))
1864            .build()
1865            .unwrap();
1866
1867        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
1868    }
1869
1870    #[test]
1871    fn test_builder_map_body_adds_processor_step() {
1872        let definition = RouteBuilder::from("timer:tick")
1873            .route_id("test-route")
1874            .map_body(|body| body)
1875            .build()
1876            .unwrap();
1877
1878        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
1879    }
1880
1881    #[test]
1882    fn test_builder_process_adds_processor_step() {
1883        let definition = RouteBuilder::from("timer:tick")
1884            .route_id("test-route")
1885            .process(|ex| async move { Ok(ex) })
1886            .build()
1887            .unwrap();
1888
1889        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
1890    }
1891
1892    #[test]
1893    fn test_builder_chain_multiple_steps() {
1894        let definition = RouteBuilder::from("timer:tick")
1895            .route_id("test-route")
1896            .set_header("source", Value::String("timer".into()))
1897            .filter(|ex| ex.input.header("source").is_some())
1898            .to("log:info")
1899            .end_filter()
1900            .to("mock:result")
1901            .build()
1902            .unwrap();
1903
1904        assert_eq!(definition.steps().len(), 3); // set_header + Filter + To("mock:result")
1905        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_))); // set_header
1906        assert!(matches!(&definition.steps()[1], BuilderStep::Filter { .. })); // filter
1907        assert!(matches!(&definition.steps()[2], BuilderStep::To(uri) if uri == "mock:result"));
1908    }
1909
1910    #[test]
1911    fn test_loop_count_builder() {
1912        use camel_api::loop_eip::LoopMode;
1913
1914        let def = RouteBuilder::from("direct:start")
1915            .route_id("loop-test")
1916            .loop_count(3)
1917            .to("mock:inside")
1918            .end_loop()
1919            .to("mock:after")
1920            .build()
1921            .unwrap();
1922
1923        assert_eq!(def.steps().len(), 2);
1924        match &def.steps()[0] {
1925            BuilderStep::Loop { config, steps } => {
1926                assert!(matches!(config.mode, LoopMode::Count(3)));
1927                assert_eq!(steps.len(), 1);
1928            }
1929            other => panic!("Expected Loop, got {:?}", other),
1930        }
1931        assert!(matches!(def.steps()[1], BuilderStep::To(_)));
1932    }
1933
1934    #[test]
1935    fn test_loop_while_builder() {
1936        use camel_api::loop_eip::LoopMode;
1937
1938        let def = RouteBuilder::from("direct:start")
1939            .route_id("loop-while-test")
1940            .loop_while(|_ex| true)
1941            .to("mock:retry")
1942            .end_loop()
1943            .build()
1944            .unwrap();
1945
1946        assert_eq!(def.steps().len(), 1);
1947        match &def.steps()[0] {
1948            BuilderStep::Loop { config, steps } => {
1949                assert!(matches!(config.mode, LoopMode::While(_)));
1950                assert_eq!(steps.len(), 1);
1951            }
1952            other => panic!("Expected Loop, got {:?}", other),
1953        }
1954    }
1955
1956    #[test]
1957    fn test_nested_loop_builder() {
1958        use camel_api::loop_eip::LoopMode;
1959
1960        let def = RouteBuilder::from("direct:start")
1961            .route_id("nested-loop-test")
1962            .loop_count(2)
1963            .to("mock:outer")
1964            .loop_count(3)
1965            .to("mock:inner")
1966            .end_loop()
1967            .end_loop()
1968            .to("mock:after")
1969            .build()
1970            .unwrap();
1971
1972        assert_eq!(def.steps().len(), 2);
1973        match &def.steps()[0] {
1974            BuilderStep::Loop { steps, .. } => {
1975                assert_eq!(steps.len(), 2);
1976                match &steps[1] {
1977                    BuilderStep::Loop {
1978                        config,
1979                        steps: inner_steps,
1980                    } => {
1981                        assert!(matches!(config.mode, LoopMode::Count(3)));
1982                        assert_eq!(inner_steps.len(), 1);
1983                    }
1984                    other => panic!("Expected nested Loop, got {:?}", other),
1985                }
1986            }
1987            other => panic!("Expected outer Loop, got {:?}", other),
1988        }
1989    }
1990
1991    // -----------------------------------------------------------------------
1992    // Processor behavior tests — exercise the real Tower services directly
1993    // -----------------------------------------------------------------------
1994
1995    #[tokio::test]
1996    async fn test_set_header_processor_works() {
1997        let mut svc = SetHeader::new(IdentityProcessor, "greeting", Value::String("hello".into()));
1998        let exchange = Exchange::new(Message::new("test"));
1999        let result = svc.call(exchange).await.unwrap();
2000        assert_eq!(
2001            result.input.header("greeting"),
2002            Some(&Value::String("hello".into()))
2003        );
2004    }
2005
2006    #[tokio::test]
2007    async fn test_filter_processor_passes() {
2008        use camel_api::BoxProcessorExt;
2009        use camel_processor::FilterService;
2010
2011        let sub = BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }));
2012        let mut svc =
2013            FilterService::new(|ex: &Exchange| ex.input.body.as_text() == Some("pass"), sub);
2014        let exchange = Exchange::new(Message::new("pass"));
2015        let result = svc.ready().await.unwrap().call(exchange).await.unwrap();
2016        assert_eq!(result.input.body.as_text(), Some("pass"));
2017    }
2018
2019    #[tokio::test]
2020    async fn test_filter_processor_blocks() {
2021        use camel_api::BoxProcessorExt;
2022        use camel_processor::FilterService;
2023
2024        let sub = BoxProcessor::from_fn(|_ex| {
2025            Box::pin(async move { Err(CamelError::ProcessorError("should not reach".into())) })
2026        });
2027        let mut svc =
2028            FilterService::new(|ex: &Exchange| ex.input.body.as_text() == Some("pass"), sub);
2029        let exchange = Exchange::new(Message::new("reject"));
2030        let result = svc.ready().await.unwrap().call(exchange).await.unwrap();
2031        assert_eq!(result.input.body.as_text(), Some("reject"));
2032    }
2033
2034    #[tokio::test]
2035    async fn test_map_body_processor_works() {
2036        let mapper = MapBody::new(IdentityProcessor, |body: Body| {
2037            if let Some(text) = body.as_text() {
2038                Body::Text(text.to_uppercase())
2039            } else {
2040                body
2041            }
2042        });
2043        let exchange = Exchange::new(Message::new("hello"));
2044        let result = mapper.oneshot(exchange).await.unwrap();
2045        assert_eq!(result.input.body.as_text(), Some("HELLO"));
2046    }
2047
2048    #[tokio::test]
2049    async fn test_process_custom_processor_works() {
2050        let processor = ProcessorFn::new(|mut ex: Exchange| async move {
2051            ex.set_property("custom", Value::Bool(true));
2052            Ok(ex)
2053        });
2054        let exchange = Exchange::new(Message::default());
2055        let result = processor.oneshot(exchange).await.unwrap();
2056        assert_eq!(result.property("custom"), Some(&Value::Bool(true)));
2057    }
2058
2059    // -----------------------------------------------------------------------
2060    // Sequential pipeline test
2061    // -----------------------------------------------------------------------
2062
2063    #[tokio::test]
2064    async fn test_compose_pipeline_runs_steps_in_order() {
2065        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2066
2067        let processors = vec![
2068            CompiledStep::Process {
2069                kind_hint: SpanKindHint::Internal,
2070                processor: BoxProcessor::new(SetHeader::new(
2071                    IdentityProcessor,
2072                    "step",
2073                    Value::String("one".into()),
2074                )),
2075                body_contract: None,
2076                lifecycle: None,
2077                label: None,
2078                to_uri: None,
2079            },
2080            CompiledStep::Process {
2081                kind_hint: SpanKindHint::Internal,
2082                processor: BoxProcessor::new(MapBody::new(IdentityProcessor, |body: Body| {
2083                    if let Some(text) = body.as_text() {
2084                        Body::Text(format!("{}-processed", text))
2085                    } else {
2086                        body
2087                    }
2088                })),
2089                body_contract: None,
2090                lifecycle: None,
2091                label: None,
2092                to_uri: None,
2093            },
2094        ];
2095
2096        let pipeline = compose_pipeline(processors, PipelineRuntimeCtx::compile_time());
2097        let exchange = Exchange::new(Message::new("hello"));
2098        let result = pipeline.oneshot(exchange).await.unwrap();
2099
2100        assert_eq!(
2101            result.input.header("step"),
2102            Some(&Value::String("one".into()))
2103        );
2104        assert_eq!(result.input.body.as_text(), Some("hello-processed"));
2105    }
2106
2107    #[tokio::test]
2108    async fn test_compose_pipeline_empty_is_identity() {
2109        use camel_core::route::{PipelineRuntimeCtx, compose_pipeline};
2110
2111        let pipeline = compose_pipeline(vec![], PipelineRuntimeCtx::compile_time());
2112        let exchange = Exchange::new(Message::new("unchanged"));
2113        let result = pipeline.oneshot(exchange).await.unwrap();
2114        assert_eq!(result.input.body.as_text(), Some("unchanged"));
2115    }
2116
2117    // -----------------------------------------------------------------------
2118    // Circuit breaker builder tests
2119    // -----------------------------------------------------------------------
2120
2121    #[test]
2122    fn test_builder_circuit_breaker_sets_config() {
2123        use camel_api::circuit_breaker::CircuitBreakerConfig;
2124
2125        let config = CircuitBreakerConfig::new().failure_threshold(5);
2126        let definition = RouteBuilder::from("timer:tick")
2127            .route_id("test-route")
2128            .circuit_breaker(config)
2129            .build()
2130            .unwrap();
2131
2132        let cb = definition
2133            .circuit_breaker_config()
2134            .expect("circuit breaker should be set");
2135        assert_eq!(cb.failure_threshold, 5);
2136    }
2137
2138    #[test]
2139    fn test_builder_circuit_breaker_with_error_handler() {
2140        use camel_api::circuit_breaker::CircuitBreakerConfig;
2141        use camel_api::error_handler::ErrorHandlerConfig;
2142
2143        let cb_config = CircuitBreakerConfig::new().failure_threshold(3);
2144        let eh_config = ErrorHandlerConfig::log_only();
2145
2146        let definition = RouteBuilder::from("timer:tick")
2147            .route_id("test-route")
2148            .to("log:info")
2149            .circuit_breaker(cb_config)
2150            .error_handler(eh_config)
2151            .build()
2152            .unwrap();
2153
2154        assert!(
2155            definition.circuit_breaker_config().is_some(),
2156            "circuit breaker config should be set"
2157        );
2158        // Route definition was built successfully with both configs.
2159    }
2160
2161    #[test]
2162    fn test_builder_on_exception_shorthand_multiple_clauses_preserve_order() {
2163        let definition = RouteBuilder::from("direct:start")
2164            .route_id("test-route")
2165            .dead_letter_channel("log:dlc")
2166            .on_exception(|e| matches!(e, CamelError::Io(_)))
2167            .retry(3)
2168            .handled_by("log:io")
2169            .end_on_exception()
2170            .on_exception(|e| matches!(e, CamelError::ProcessorError(_)))
2171            .retry(1)
2172            .end_on_exception()
2173            .to("mock:out")
2174            .build()
2175            .expect("route should build");
2176
2177        let cfg = definition
2178            .error_handler_config()
2179            .expect("error handler should be set");
2180        assert_eq!(cfg.policies.len(), 2);
2181        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:dlc"));
2182        assert_eq!(
2183            cfg.policies[0].retry.as_ref().map(|p| p.max_attempts),
2184            Some(3)
2185        );
2186        assert_eq!(cfg.policies[0].handled_by.as_deref(), Some("log:io"));
2187        assert_eq!(
2188            cfg.policies[1].retry.as_ref().map(|p| p.max_attempts),
2189            Some(1)
2190        );
2191    }
2192
2193    #[test]
2194    fn test_builder_on_exception_mixed_mode_rejected() {
2195        let result = RouteBuilder::from("direct:start")
2196            .route_id("test-route")
2197            .error_handler(ErrorHandlerConfig::log_only())
2198            .on_exception(|_e| true)
2199            .end_on_exception()
2200            .to("mock:out")
2201            .build();
2202
2203        let err = result.err().expect("mixed mode should fail with an error");
2204
2205        assert!(
2206            format!("{err}").contains("mixed error handler modes"),
2207            "unexpected error: {err}"
2208        );
2209    }
2210
2211    #[test]
2212    fn test_builder_on_exception_backoff_and_jitter_without_retry_noop() {
2213        let definition = RouteBuilder::from("direct:start")
2214            .route_id("test-route")
2215            .on_exception(|_e| true)
2216            .with_backoff(Duration::from_millis(5), 3.0, Duration::from_millis(100))
2217            .with_jitter(0.5)
2218            .end_on_exception()
2219            .to("mock:out")
2220            .build()
2221            .expect("route should build");
2222
2223        let cfg = definition
2224            .error_handler_config()
2225            .expect("error handler should be set");
2226        assert_eq!(cfg.policies.len(), 1);
2227        assert!(cfg.policies[0].retry.is_none());
2228    }
2229
2230    #[test]
2231    fn test_builder_dead_letter_channel_without_on_exception_sets_dlc() {
2232        let definition = RouteBuilder::from("direct:start")
2233            .route_id("test-route")
2234            .dead_letter_channel("log:dlc")
2235            .to("mock:out")
2236            .build()
2237            .expect("route should build");
2238
2239        let cfg = definition
2240            .error_handler_config()
2241            .expect("error handler should be set");
2242        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:dlc"));
2243        assert!(cfg.policies.is_empty());
2244    }
2245
2246    #[test]
2247    fn test_builder_dead_letter_channel_called_twice_uses_latest_and_keeps_policies() {
2248        let definition = RouteBuilder::from("direct:start")
2249            .route_id("test-route")
2250            .dead_letter_channel("log:first")
2251            .on_exception(|e| matches!(e, CamelError::Io(_)))
2252            .retry(2)
2253            .end_on_exception()
2254            .dead_letter_channel("log:second")
2255            .to("mock:out")
2256            .build()
2257            .expect("route should build");
2258
2259        let cfg = definition
2260            .error_handler_config()
2261            .expect("error handler should be set");
2262        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:second"));
2263        assert_eq!(cfg.policies.len(), 1);
2264        assert_eq!(
2265            cfg.policies[0].retry.as_ref().map(|p| p.max_attempts),
2266            Some(2)
2267        );
2268    }
2269
2270    #[test]
2271    fn test_builder_on_exception_without_dlc_defaults_to_log_only() {
2272        let definition = RouteBuilder::from("direct:start")
2273            .route_id("test-route")
2274            .on_exception(|e| matches!(e, CamelError::ProcessorError(_)))
2275            .retry(1)
2276            .end_on_exception()
2277            .to("mock:out")
2278            .build()
2279            .expect("route should build");
2280
2281        let cfg = definition
2282            .error_handler_config()
2283            .expect("error handler should be set");
2284        assert!(cfg.dlc_uri.is_none());
2285        assert_eq!(cfg.policies.len(), 1);
2286    }
2287
2288    #[test]
2289    fn test_builder_error_handler_explicit_overwrite_stays_explicit_mode() {
2290        let first = ErrorHandlerConfig::dead_letter_channel("log:first");
2291        let second = ErrorHandlerConfig::dead_letter_channel("log:second");
2292
2293        let definition = RouteBuilder::from("direct:start")
2294            .route_id("test-route")
2295            .error_handler(first)
2296            .error_handler(second)
2297            .to("mock:out")
2298            .build()
2299            .expect("route should build");
2300
2301        let cfg = definition
2302            .error_handler_config()
2303            .expect("error handler should be set");
2304        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:second"));
2305    }
2306
2307    // --- Splitter builder tests ---
2308
2309    #[test]
2310    fn test_split_builder_typestate() {
2311        use camel_api::splitter::{SplitterConfig, split_body_lines};
2312
2313        // .split() returns SplitBuilder, .end_split() returns RouteBuilder
2314        let definition = RouteBuilder::from("timer:test?period=1000")
2315            .route_id("test-route")
2316            .split(SplitterConfig::new(split_body_lines()))
2317            .to("mock:per-fragment")
2318            .end_split()
2319            .to("mock:final")
2320            .build()
2321            .unwrap();
2322
2323        // Should have 2 top-level steps: Split + To("mock:final")
2324        assert_eq!(definition.steps().len(), 2);
2325    }
2326
2327    #[test]
2328    fn test_split_builder_steps_collected() {
2329        use camel_api::splitter::{SplitterConfig, split_body_lines};
2330
2331        let definition = RouteBuilder::from("timer:test?period=1000")
2332            .route_id("test-route")
2333            .split(SplitterConfig::new(split_body_lines()))
2334            .set_header("fragment", Value::String("yes".into()))
2335            .to("mock:per-fragment")
2336            .end_split()
2337            .build()
2338            .unwrap();
2339
2340        // Should have 1 top-level step: Split (containing 2 sub-steps)
2341        assert_eq!(definition.steps().len(), 1);
2342        match &definition.steps()[0] {
2343            BuilderStep::Split { steps, .. } => {
2344                assert_eq!(steps.len(), 2); // SetHeader + To
2345            }
2346            other => panic!("Expected Split, got {:?}", other),
2347        }
2348    }
2349
2350    #[test]
2351    fn test_split_builder_config_propagated() {
2352        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
2353
2354        let definition = RouteBuilder::from("timer:test?period=1000")
2355            .route_id("test-route")
2356            .split(
2357                SplitterConfig::new(split_body_lines())
2358                    .parallel(true)
2359                    .parallel_limit(4)
2360                    .aggregation(AggregationStrategy::CollectAll),
2361            )
2362            .to("mock:per-fragment")
2363            .end_split()
2364            .build()
2365            .unwrap();
2366
2367        match &definition.steps()[0] {
2368            BuilderStep::Split { config, .. } => {
2369                assert!(config.parallel);
2370                assert_eq!(config.parallel_limit, Some(4));
2371                assert!(matches!(
2372                    config.aggregation,
2373                    AggregationStrategy::CollectAll
2374                ));
2375            }
2376            other => panic!("Expected Split, got {:?}", other),
2377        }
2378    }
2379
2380    #[test]
2381    fn test_aggregate_builder_adds_step() {
2382        use camel_api::aggregator::AggregatorConfig;
2383        use camel_core::route::BuilderStep;
2384
2385        let definition = RouteBuilder::from("timer:tick")
2386            .route_id("test-route")
2387            .aggregate(
2388                AggregatorConfig::correlate_by("key")
2389                    .complete_when_size(2)
2390                    .build()
2391                    .unwrap(),
2392            )
2393            .build()
2394            .unwrap();
2395
2396        assert_eq!(definition.steps().len(), 1);
2397        assert!(matches!(
2398            definition.steps()[0],
2399            BuilderStep::Aggregate { .. }
2400        ));
2401    }
2402
2403    #[test]
2404    fn test_aggregate_in_split_builder() {
2405        use camel_api::aggregator::AggregatorConfig;
2406        use camel_api::splitter::{SplitterConfig, split_body_lines};
2407        use camel_core::route::BuilderStep;
2408
2409        let definition = RouteBuilder::from("timer:tick")
2410            .route_id("test-route")
2411            .split(SplitterConfig::new(split_body_lines()))
2412            .aggregate(
2413                AggregatorConfig::correlate_by("key")
2414                    .complete_when_size(1)
2415                    .build()
2416                    .unwrap(),
2417            )
2418            .end_split()
2419            .build()
2420            .unwrap();
2421
2422        assert_eq!(definition.steps().len(), 1);
2423        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
2424            assert!(matches!(steps[0], BuilderStep::Aggregate { .. }));
2425        } else {
2426            panic!("expected Split step");
2427        }
2428    }
2429
2430    // ── set_body / set_body_fn / set_header_fn builder tests ────────────────────
2431
2432    #[test]
2433    fn test_builder_set_body_static_adds_processor() {
2434        let definition = RouteBuilder::from("timer:tick")
2435            .route_id("test-route")
2436            .set_body("fixed")
2437            .build()
2438            .unwrap();
2439        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
2440    }
2441
2442    #[test]
2443    fn test_builder_set_body_fn_adds_processor() {
2444        let definition = RouteBuilder::from("timer:tick")
2445            .route_id("test-route")
2446            .set_body_fn(|_ex: &Exchange| Body::Text("dynamic".into()))
2447            .build()
2448            .unwrap();
2449        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
2450    }
2451
2452    #[test]
2453    fn transform_alias_produces_same_as_set_body() {
2454        let route_transform = RouteBuilder::from("timer:tick")
2455            .route_id("test-route")
2456            .transform("hello")
2457            .build()
2458            .unwrap();
2459
2460        let route_set_body = RouteBuilder::from("timer:tick")
2461            .route_id("test-route")
2462            .set_body("hello")
2463            .build()
2464            .unwrap();
2465
2466        assert_eq!(route_transform.steps().len(), route_set_body.steps().len());
2467    }
2468
2469    #[test]
2470    fn test_builder_set_header_fn_adds_processor() {
2471        let definition = RouteBuilder::from("timer:tick")
2472            .route_id("test-route")
2473            .set_header_fn("k", |_ex: &Exchange| Value::String("v".into()))
2474            .build()
2475            .unwrap();
2476        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
2477    }
2478
2479    #[tokio::test]
2480    async fn test_set_body_static_processor_works() {
2481        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2482        let def = RouteBuilder::from("t:t")
2483            .route_id("test-route")
2484            .set_body("replaced")
2485            .build()
2486            .unwrap();
2487        let pipeline = compose_pipeline(
2488            def.steps()
2489                .iter()
2490                .filter_map(|s| {
2491                    if let BuilderStep::Processor(op) = s {
2492                        Some(op.0.clone())
2493                    } else {
2494                        None
2495                    }
2496                })
2497                .map(|p| CompiledStep::Process {
2498                    kind_hint: SpanKindHint::Internal,
2499                    processor: p,
2500                    body_contract: None,
2501                    lifecycle: None,
2502                    label: None,
2503                    to_uri: None,
2504                })
2505                .collect(),
2506            PipelineRuntimeCtx::compile_time(),
2507        );
2508        let exchange = Exchange::new(Message::new("original"));
2509        let result = pipeline.oneshot(exchange).await.unwrap();
2510        assert_eq!(result.input.body.as_text(), Some("replaced"));
2511    }
2512
2513    #[tokio::test]
2514    async fn test_set_body_fn_processor_works() {
2515        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2516        let def = RouteBuilder::from("t:t")
2517            .route_id("test-route")
2518            .set_body_fn(|ex: &Exchange| {
2519                Body::Text(ex.input.body.as_text().unwrap_or("").to_uppercase())
2520            })
2521            .build()
2522            .unwrap();
2523        let pipeline = compose_pipeline(
2524            def.steps()
2525                .iter()
2526                .filter_map(|s| {
2527                    if let BuilderStep::Processor(op) = s {
2528                        Some(op.0.clone())
2529                    } else {
2530                        None
2531                    }
2532                })
2533                .map(|p| CompiledStep::Process {
2534                    kind_hint: SpanKindHint::Internal,
2535                    processor: p,
2536                    body_contract: None,
2537                    lifecycle: None,
2538                    label: None,
2539                    to_uri: None,
2540                })
2541                .collect(),
2542            PipelineRuntimeCtx::compile_time(),
2543        );
2544        let exchange = Exchange::new(Message::new("hello"));
2545        let result = pipeline.oneshot(exchange).await.unwrap();
2546        assert_eq!(result.input.body.as_text(), Some("HELLO"));
2547    }
2548
2549    #[tokio::test]
2550    async fn test_set_header_fn_processor_works() {
2551        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2552        let def = RouteBuilder::from("t:t")
2553            .route_id("test-route")
2554            .set_header_fn("echo", |ex: &Exchange| {
2555                ex.input
2556                    .body
2557                    .as_text()
2558                    .map(|t| Value::String(t.into()))
2559                    .unwrap_or(Value::Null)
2560            })
2561            .build()
2562            .unwrap();
2563        let pipeline = compose_pipeline(
2564            def.steps()
2565                .iter()
2566                .filter_map(|s| {
2567                    if let BuilderStep::Processor(op) = s {
2568                        Some(op.0.clone())
2569                    } else {
2570                        None
2571                    }
2572                })
2573                .map(|p| CompiledStep::Process {
2574                    kind_hint: SpanKindHint::Internal,
2575                    processor: p,
2576                    body_contract: None,
2577                    lifecycle: None,
2578                    label: None,
2579                    to_uri: None,
2580                })
2581                .collect(),
2582            PipelineRuntimeCtx::compile_time(),
2583        );
2584        let exchange = Exchange::new(Message::new("ping"));
2585        let result = pipeline.oneshot(exchange).await.unwrap();
2586        assert_eq!(
2587            result.input.header("echo"),
2588            Some(&Value::String("ping".into()))
2589        );
2590    }
2591
2592    // ── FilterBuilder typestate tests ─────────────────────────────────────
2593
2594    #[test]
2595    fn test_filter_builder_typestate() {
2596        let result = RouteBuilder::from("timer:tick?period=50&repeatCount=1")
2597            .route_id("test-route")
2598            .filter(|_ex| true)
2599            .to("mock:inner")
2600            .end_filter()
2601            .to("mock:outer")
2602            .build();
2603        assert!(result.is_ok());
2604    }
2605
2606    #[test]
2607    fn test_filter_builder_steps_collected() {
2608        let definition = RouteBuilder::from("timer:tick?period=50&repeatCount=1")
2609            .route_id("test-route")
2610            .filter(|_ex| true)
2611            .to("mock:inner")
2612            .end_filter()
2613            .build()
2614            .unwrap();
2615
2616        assert_eq!(definition.steps().len(), 1);
2617        assert!(matches!(&definition.steps()[0], BuilderStep::Filter { .. }));
2618    }
2619
2620    #[test]
2621    fn test_wire_tap_builder_adds_step() {
2622        let definition = RouteBuilder::from("timer:tick")
2623            .route_id("test-route")
2624            .wire_tap("mock:tap")
2625            .to("mock:result")
2626            .build()
2627            .unwrap();
2628
2629        assert_eq!(definition.steps().len(), 2);
2630        assert!(
2631            matches!(&definition.steps()[0], BuilderStep::WireTap { uri } if uri == "mock:tap")
2632        );
2633        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:result"));
2634    }
2635
2636    // ── MulticastBuilder typestate tests ─────────────────────────────────────
2637
2638    #[test]
2639    fn test_multicast_builder_typestate() {
2640        let definition = RouteBuilder::from("timer:tick")
2641            .route_id("test-route")
2642            .multicast()
2643            .to("direct:a")
2644            .to("direct:b")
2645            .end_multicast()
2646            .to("mock:result")
2647            .build()
2648            .unwrap();
2649
2650        assert_eq!(definition.steps().len(), 2); // Multicast + To("mock:result")
2651    }
2652
2653    #[test]
2654    fn test_multicast_builder_steps_collected() {
2655        let definition = RouteBuilder::from("timer:tick")
2656            .route_id("test-route")
2657            .multicast()
2658            .to("direct:a")
2659            .to("direct:b")
2660            .end_multicast()
2661            .build()
2662            .unwrap();
2663
2664        match &definition.steps()[0] {
2665            BuilderStep::Multicast { steps, .. } => {
2666                assert_eq!(steps.len(), 2);
2667            }
2668            other => panic!("Expected Multicast, got {:?}", other),
2669        }
2670    }
2671
2672    // ── Concurrency builder tests ─────────────────────────────────────
2673
2674    #[test]
2675    fn test_builder_concurrent_sets_concurrency() {
2676        use camel_component_api::ConcurrencyModel;
2677
2678        let definition = RouteBuilder::from("http://0.0.0.0:8080/test")
2679            .route_id("test-route")
2680            .concurrent(16)
2681            .to("log:info")
2682            .build()
2683            .unwrap();
2684
2685        assert_eq!(
2686            definition.concurrency_override(),
2687            Some(&ConcurrencyModel::Concurrent { max: Some(16) })
2688        );
2689    }
2690
2691    #[test]
2692    fn test_builder_concurrent_zero_means_unbounded() {
2693        use camel_component_api::ConcurrencyModel;
2694
2695        let definition = RouteBuilder::from("http://0.0.0.0:8080/test")
2696            .route_id("test-route")
2697            .concurrent(0)
2698            .to("log:info")
2699            .build()
2700            .unwrap();
2701
2702        assert_eq!(
2703            definition.concurrency_override(),
2704            Some(&ConcurrencyModel::Concurrent { max: None })
2705        );
2706    }
2707
2708    #[test]
2709    fn test_builder_sequential_sets_concurrency() {
2710        use camel_component_api::ConcurrencyModel;
2711
2712        let definition = RouteBuilder::from("http://0.0.0.0:8080/test")
2713            .route_id("test-route")
2714            .sequential()
2715            .to("log:info")
2716            .build()
2717            .unwrap();
2718
2719        assert_eq!(
2720            definition.concurrency_override(),
2721            Some(&ConcurrencyModel::Sequential)
2722        );
2723    }
2724
2725    #[test]
2726    fn test_builder_default_concurrency_is_none() {
2727        let definition = RouteBuilder::from("timer:tick")
2728            .route_id("test-route")
2729            .to("log:info")
2730            .build()
2731            .unwrap();
2732
2733        assert_eq!(definition.concurrency_override(), None);
2734    }
2735
2736    // ── Route lifecycle builder tests ─────────────────────────────────────
2737
2738    #[test]
2739    fn test_builder_route_id_sets_id() {
2740        let definition = RouteBuilder::from("timer:tick")
2741            .route_id("my-route")
2742            .build()
2743            .unwrap();
2744
2745        assert_eq!(definition.route_id(), "my-route");
2746    }
2747
2748    #[test]
2749    fn test_build_without_route_id_fails() {
2750        let result = RouteBuilder::from("timer:tick?period=1000")
2751            .to("log:info")
2752            .build();
2753        let err = match result {
2754            Err(e) => e.to_string(),
2755            Ok(_) => panic!("build() should fail without route_id"),
2756        };
2757        assert!(
2758            err.contains("route_id"),
2759            "error should mention route_id, got: {}",
2760            err
2761        );
2762    }
2763
2764    #[test]
2765    fn test_builder_empty_route_id_rejected() {
2766        let result = RouteBuilder::from("timer:tick").route_id("").build();
2767        let err = result.err().expect("empty route_id should be rejected");
2768        assert!(matches!(err, CamelError::RouteError(_)));
2769    }
2770
2771    #[test]
2772    fn test_builder_whitespace_route_id_rejected() {
2773        let result = RouteBuilder::from("timer:tick").route_id("   ").build();
2774        assert!(result.is_err());
2775    }
2776
2777    #[test]
2778    fn test_builder_auto_startup_false() {
2779        let definition = RouteBuilder::from("timer:tick")
2780            .route_id("test-route")
2781            .auto_startup(false)
2782            .build()
2783            .unwrap();
2784
2785        assert!(!definition.auto_startup());
2786    }
2787
2788    #[test]
2789    fn test_builder_startup_order_custom() {
2790        let definition = RouteBuilder::from("timer:tick")
2791            .route_id("test-route")
2792            .startup_order(50)
2793            .build()
2794            .unwrap();
2795
2796        assert_eq!(definition.startup_order(), 50);
2797    }
2798
2799    #[test]
2800    fn test_builder_defaults() {
2801        let definition = RouteBuilder::from("timer:tick")
2802            .route_id("test-route")
2803            .build()
2804            .unwrap();
2805
2806        assert_eq!(definition.route_id(), "test-route");
2807        assert!(definition.auto_startup());
2808        assert_eq!(definition.startup_order(), 1000);
2809    }
2810
2811    // ── Choice typestate tests ──────────────────────────────────────────────────
2812
2813    #[test]
2814    fn test_choice_builder_single_when() {
2815        let definition = RouteBuilder::from("timer:tick")
2816            .route_id("test-route")
2817            .choice()
2818            .when(|ex: &Exchange| ex.input.header("type").is_some())
2819            .to("mock:typed")
2820            .end_when()
2821            .end_choice()
2822            .build()
2823            .unwrap();
2824        assert_eq!(definition.steps().len(), 1);
2825        assert!(
2826            matches!(&definition.steps()[0], BuilderStep::Choice { whens, otherwise }
2827            if whens.len() == 1 && otherwise.is_none())
2828        );
2829    }
2830
2831    #[test]
2832    fn test_choice_builder_when_otherwise() {
2833        let definition = RouteBuilder::from("timer:tick")
2834            .route_id("test-route")
2835            .choice()
2836            .when(|ex: &Exchange| ex.input.header("a").is_some())
2837            .to("mock:a")
2838            .end_when()
2839            .otherwise()
2840            .to("mock:fallback")
2841            .end_otherwise()
2842            .end_choice()
2843            .build()
2844            .unwrap();
2845        assert!(
2846            matches!(&definition.steps()[0], BuilderStep::Choice { whens, otherwise }
2847            if whens.len() == 1 && otherwise.is_some())
2848        );
2849    }
2850
2851    #[test]
2852    fn test_choice_builder_multiple_whens() {
2853        let definition = RouteBuilder::from("timer:tick")
2854            .route_id("test-route")
2855            .choice()
2856            .when(|ex: &Exchange| ex.input.header("a").is_some())
2857            .to("mock:a")
2858            .end_when()
2859            .when(|ex: &Exchange| ex.input.header("b").is_some())
2860            .to("mock:b")
2861            .end_when()
2862            .end_choice()
2863            .build()
2864            .unwrap();
2865        assert!(
2866            matches!(&definition.steps()[0], BuilderStep::Choice { whens, .. }
2867            if whens.len() == 2)
2868        );
2869    }
2870
2871    #[test]
2872    fn test_choice_step_after_choice() {
2873        // Steps after end_choice() are added to the outer pipeline, not inside choice.
2874        let definition = RouteBuilder::from("timer:tick")
2875            .route_id("test-route")
2876            .choice()
2877            .when(|_ex: &Exchange| true)
2878            .to("mock:inner")
2879            .end_when()
2880            .end_choice()
2881            .to("mock:outer") // must be step[1], not inside choice
2882            .build()
2883            .unwrap();
2884        assert_eq!(definition.steps().len(), 2);
2885        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
2886    }
2887
2888    // ── Throttle typestate tests ──────────────────────────────────────────────────
2889
2890    #[test]
2891    fn test_throttle_builder_typestate() {
2892        let definition = RouteBuilder::from("timer:tick")
2893            .route_id("test-route")
2894            .throttle(10, std::time::Duration::from_secs(1))
2895            .to("mock:result")
2896            .end_throttle()
2897            .build()
2898            .unwrap();
2899
2900        assert_eq!(definition.steps().len(), 1);
2901        assert!(matches!(
2902            &definition.steps()[0],
2903            BuilderStep::Throttle { .. }
2904        ));
2905    }
2906
2907    #[test]
2908    fn test_throttle_builder_with_strategy() {
2909        let definition = RouteBuilder::from("timer:tick")
2910            .route_id("test-route")
2911            .throttle(10, std::time::Duration::from_secs(1))
2912            .strategy(ThrottleStrategy::Reject)
2913            .to("mock:result")
2914            .end_throttle()
2915            .build()
2916            .unwrap();
2917
2918        if let BuilderStep::Throttle { config, .. } = &definition.steps()[0] {
2919            assert_eq!(config.strategy, ThrottleStrategy::Reject);
2920        } else {
2921            panic!("Expected Throttle step");
2922        }
2923    }
2924
2925    #[test]
2926    fn test_throttle_builder_steps_collected() {
2927        let definition = RouteBuilder::from("timer:tick")
2928            .route_id("test-route")
2929            .throttle(5, std::time::Duration::from_secs(1))
2930            .set_header("throttled", Value::Bool(true))
2931            .to("mock:throttled")
2932            .end_throttle()
2933            .build()
2934            .unwrap();
2935
2936        match &definition.steps()[0] {
2937            BuilderStep::Throttle { steps, .. } => {
2938                assert_eq!(steps.len(), 2); // SetHeader + To
2939            }
2940            other => panic!("Expected Throttle, got {:?}", other),
2941        }
2942    }
2943
2944    #[test]
2945    fn test_throttle_step_after_throttle() {
2946        // Steps after end_throttle() are added to the outer pipeline, not inside throttle.
2947        let definition = RouteBuilder::from("timer:tick")
2948            .route_id("test-route")
2949            .throttle(10, std::time::Duration::from_secs(1))
2950            .to("mock:inner")
2951            .end_throttle()
2952            .to("mock:outer")
2953            .build()
2954            .unwrap();
2955
2956        assert_eq!(definition.steps().len(), 2);
2957        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
2958    }
2959
2960    // ── LoadBalance typestate tests ──────────────────────────────────────────────────
2961
2962    #[test]
2963    fn test_load_balance_builder_typestate() {
2964        let definition = RouteBuilder::from("timer:tick")
2965            .route_id("test-route")
2966            .load_balance()
2967            .round_robin()
2968            .to("mock:a")
2969            .to("mock:b")
2970            .end_load_balance()
2971            .build()
2972            .unwrap();
2973
2974        assert_eq!(definition.steps().len(), 1);
2975        assert!(matches!(
2976            &definition.steps()[0],
2977            BuilderStep::LoadBalance { .. }
2978        ));
2979    }
2980
2981    #[test]
2982    fn test_load_balance_builder_with_strategy() {
2983        let definition = RouteBuilder::from("timer:tick")
2984            .route_id("test-route")
2985            .load_balance()
2986            .random()
2987            .to("mock:result")
2988            .end_load_balance()
2989            .build()
2990            .unwrap();
2991
2992        if let BuilderStep::LoadBalance { config, .. } = &definition.steps()[0] {
2993            assert_eq!(config.strategy, LoadBalanceStrategy::Random);
2994        } else {
2995            panic!("Expected LoadBalance step");
2996        }
2997    }
2998
2999    #[test]
3000    fn test_load_balance_builder_steps_collected() {
3001        let definition = RouteBuilder::from("timer:tick")
3002            .route_id("test-route")
3003            .load_balance()
3004            .set_header("lb", Value::Bool(true))
3005            .to("mock:a")
3006            .end_load_balance()
3007            .build()
3008            .unwrap();
3009
3010        match &definition.steps()[0] {
3011            BuilderStep::LoadBalance { steps, .. } => {
3012                assert_eq!(steps.len(), 2); // SetHeader + To
3013            }
3014            other => panic!("Expected LoadBalance, got {:?}", other),
3015        }
3016    }
3017
3018    #[test]
3019    fn test_load_balance_step_after_load_balance() {
3020        // Steps after end_load_balance() are added to the outer pipeline, not inside load_balance.
3021        let definition = RouteBuilder::from("timer:tick")
3022            .route_id("test-route")
3023            .load_balance()
3024            .to("mock:inner")
3025            .end_load_balance()
3026            .to("mock:outer")
3027            .build()
3028            .unwrap();
3029
3030        assert_eq!(definition.steps().len(), 2);
3031        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
3032    }
3033
3034    // ── DynamicRouter typestate tests ──────────────────────────────────────────────────
3035
3036    #[test]
3037    fn test_dynamic_router_builder() {
3038        let definition = RouteBuilder::from("timer:tick")
3039            .route_id("test-route")
3040            .dynamic_router(Arc::new(|_| Some("mock:result".to_string())))
3041            .build()
3042            .unwrap();
3043
3044        assert_eq!(definition.steps().len(), 1);
3045        assert!(matches!(
3046            &definition.steps()[0],
3047            BuilderStep::DynamicRouter { .. }
3048        ));
3049    }
3050
3051    #[test]
3052    fn test_dynamic_router_builder_with_config() {
3053        let config = DynamicRouterConfig::new(Arc::new(|_| Some("mock:a".to_string())))
3054            .max_iterations(100)
3055            .cache_size(500);
3056
3057        let definition = RouteBuilder::from("timer:tick")
3058            .route_id("test-route")
3059            .dynamic_router_with_config(config)
3060            .build()
3061            .unwrap();
3062
3063        assert_eq!(definition.steps().len(), 1);
3064        if let BuilderStep::DynamicRouter { config } = &definition.steps()[0] {
3065            assert_eq!(config.max_iterations, 100);
3066            assert_eq!(config.cache_size, 500);
3067        } else {
3068            panic!("Expected DynamicRouter step");
3069        }
3070    }
3071
3072    #[test]
3073    fn test_dynamic_router_step_after_router() {
3074        // Steps after dynamic_router() are added to the outer pipeline.
3075        let definition = RouteBuilder::from("timer:tick")
3076            .route_id("test-route")
3077            .dynamic_router(Arc::new(|_| Some("mock:inner".to_string())))
3078            .to("mock:outer")
3079            .build()
3080            .unwrap();
3081
3082        assert_eq!(definition.steps().len(), 2);
3083        assert!(matches!(
3084            &definition.steps()[0],
3085            BuilderStep::DynamicRouter { .. }
3086        ));
3087        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
3088    }
3089
3090    #[test]
3091    fn routing_slip_builder_creates_step() {
3092        use camel_api::RoutingSlipExpression;
3093
3094        let expression: RoutingSlipExpression = Arc::new(|_| Some("direct:a,direct:b".to_string()));
3095
3096        let route = RouteBuilder::from("direct:start")
3097            .route_id("routing-slip-test")
3098            .routing_slip(expression)
3099            .build()
3100            .unwrap();
3101
3102        assert!(
3103            matches!(route.steps()[0], BuilderStep::RoutingSlip { .. }),
3104            "Expected RoutingSlip step"
3105        );
3106    }
3107
3108    #[test]
3109    fn routing_slip_with_config_builder_creates_step() {
3110        use camel_api::RoutingSlipConfig;
3111
3112        let config = RoutingSlipConfig::new(Arc::new(|_| Some("mock:a".to_string())))
3113            .uri_delimiter("|")
3114            .cache_size(50)
3115            .ignore_invalid_endpoints(true);
3116
3117        let route = RouteBuilder::from("direct:start")
3118            .route_id("routing-slip-config-test")
3119            .routing_slip_with_config(config)
3120            .build()
3121            .unwrap();
3122
3123        if let BuilderStep::RoutingSlip { config } = &route.steps()[0] {
3124            assert_eq!(config.uri_delimiter, "|");
3125            assert_eq!(config.cache_size, 50);
3126            assert!(config.ignore_invalid_endpoints);
3127        } else {
3128            panic!("Expected RoutingSlip step");
3129        }
3130    }
3131
3132    #[test]
3133    fn test_builder_marshal_adds_processor_step() {
3134        let definition = RouteBuilder::from("timer:tick")
3135            .route_id("test-route")
3136            .marshal("json")
3137            .unwrap()
3138            .build()
3139            .unwrap();
3140        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3141    }
3142
3143    #[test]
3144    fn test_builder_unmarshal_adds_processor_step() {
3145        let definition = RouteBuilder::from("timer:tick")
3146            .route_id("test-route")
3147            .unmarshal("json")
3148            .unwrap()
3149            .build()
3150            .unwrap();
3151        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3152    }
3153
3154    #[test]
3155    fn test_builder_stream_cache_adds_processor_step() {
3156        let definition = RouteBuilder::from("timer:tick")
3157            .route_id("test-route")
3158            .stream_cache(1024)
3159            .build()
3160            .unwrap();
3161        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3162    }
3163
3164    #[test]
3165    fn validate_adds_validate_step() {
3166        let def = RouteBuilder::from("direct:in")
3167            .route_id("test")
3168            .validate("schemas/order.xsd")
3169            .build()
3170            .unwrap();
3171        let steps = def.steps();
3172        assert_eq!(steps.len(), 1);
3173        assert!(
3174            matches!(&steps[0], BuilderStep::Validate { predicate } if predicate.language == "simple" && predicate.source == "schemas/order.xsd"),
3175            "got: {:?}",
3176            steps[0]
3177        );
3178    }
3179
3180    #[test]
3181    fn test_builder_marshal_returns_err_for_unknown_format() {
3182        let result = RouteBuilder::from("timer:tick")
3183            .route_id("test-route")
3184            .marshal("protobuf");
3185        let err = match result {
3186            Err(e) => e,
3187            Ok(_) => panic!("marshal with unknown format should return Err"),
3188        };
3189        let msg = err.to_string();
3190        assert!(
3191            msg.contains("unknown data format"),
3192            "error should mention unknown format, got: {msg}"
3193        );
3194        assert!(
3195            msg.contains("protobuf"),
3196            "error should mention format name, got: {msg}"
3197        );
3198    }
3199
3200    #[test]
3201    fn test_builder_unmarshal_returns_err_for_unknown_format() {
3202        let result = RouteBuilder::from("timer:tick")
3203            .route_id("test-route")
3204            .unmarshal("protobuf");
3205        let err = match result {
3206            Err(e) => e,
3207            Ok(_) => panic!("unmarshal with unknown format should return Err"),
3208        };
3209        let msg = err.to_string();
3210        assert!(
3211            msg.contains("unknown data format"),
3212            "error should mention unknown format, got: {msg}"
3213        );
3214        assert!(
3215            msg.contains("protobuf"),
3216            "error should mention format name, got: {msg}"
3217        );
3218    }
3219
3220    #[test]
3221    fn test_builder_recipient_list_creates_step() {
3222        let route = RouteBuilder::from("direct:start")
3223            .route_id("recipient-list-test")
3224            .recipient_list(Arc::new(|_| "direct:a,direct:b".to_string()))
3225            .build()
3226            .unwrap();
3227
3228        assert!(matches!(
3229            &route.steps()[0],
3230            BuilderStep::RecipientList { .. }
3231        ));
3232    }
3233
3234    #[test]
3235    fn test_builder_recipient_list_with_config_creates_step() {
3236        let config = RecipientListConfig::new(Arc::new(|_| "mock:a".to_string()));
3237
3238        let route = RouteBuilder::from("direct:start")
3239            .route_id("recipient-list-config-test")
3240            .recipient_list_with_config(config)
3241            .build()
3242            .unwrap();
3243
3244        assert!(matches!(
3245            &route.steps()[0],
3246            BuilderStep::RecipientList { .. }
3247        ));
3248    }
3249
3250    #[test]
3251    fn test_builder_script_adds_script_step() {
3252        let route = RouteBuilder::from("direct:start")
3253            .route_id("script-test")
3254            .script("rhai", "headers[\"x\"] = \"y\"")
3255            .build()
3256            .unwrap();
3257
3258        assert!(matches!(
3259            &route.steps()[0],
3260            BuilderStep::Script { language, script }
3261            if language == "rhai" && script == "headers[\"x\"] = \"y\""
3262        ));
3263    }
3264
3265    #[test]
3266    fn test_builder_delay_and_delay_with_header_add_steps() {
3267        let route = RouteBuilder::from("direct:start")
3268            .route_id("delay-test")
3269            .delay(Duration::from_millis(250))
3270            .delay_with_header(Duration::from_millis(500), "x-delay")
3271            .build()
3272            .unwrap();
3273
3274        assert_eq!(route.steps().len(), 2);
3275        assert!(matches!(&route.steps()[0], BuilderStep::Delay { .. }));
3276        assert!(matches!(&route.steps()[1], BuilderStep::Delay { .. }));
3277    }
3278
3279    #[test]
3280    fn test_builder_log_and_stop_add_steps_in_order() {
3281        let route = RouteBuilder::from("direct:start")
3282            .route_id("log-stop-test")
3283            .log("hello", LogLevel::Info)
3284            .stop()
3285            .to("mock:after")
3286            .build()
3287            .unwrap();
3288
3289        assert_eq!(route.steps().len(), 3);
3290        assert!(matches!(
3291            &route.steps()[0],
3292            BuilderStep::Log { message, .. } if message == "hello"
3293        ));
3294        assert!(matches!(&route.steps()[1], BuilderStep::Stop));
3295        assert!(matches!(&route.steps()[2], BuilderStep::To(uri) if uri == "mock:after"));
3296    }
3297
3298    #[test]
3299    fn test_builder_stream_cache_default_adds_processor_step() {
3300        let route = RouteBuilder::from("direct:start")
3301            .route_id("stream-cache-default-test")
3302            .stream_cache_default()
3303            .build()
3304            .unwrap();
3305
3306        assert!(matches!(&route.steps()[0], BuilderStep::Processor(_)));
3307    }
3308
3309    #[test]
3310    fn test_validate_creates_validate_step_with_expression() {
3311        let route = RouteBuilder::from("direct:in")
3312            .route_id("validate-prefix-test")
3313            .validate("${body.size()} > 0")
3314            .build()
3315            .unwrap();
3316
3317        assert!(matches!(
3318            &route.steps()[0],
3319            BuilderStep::Validate { predicate } if predicate.language == "simple" && predicate.source == "${body.size()} > 0"
3320        ));
3321    }
3322
3323    #[test]
3324    fn test_load_balance_builder_weighted_failover_config() {
3325        let route = RouteBuilder::from("direct:start")
3326            .route_id("lb-weighted-failover")
3327            .load_balance()
3328            .weighted(vec![
3329                ("direct:a".to_string(), 3),
3330                ("direct:b".to_string(), 1),
3331            ])
3332            .failover()
3333            .to("mock:result")
3334            .end_load_balance()
3335            .build()
3336            .unwrap();
3337
3338        if let BuilderStep::LoadBalance { config, .. } = &route.steps()[0] {
3339            assert_eq!(config.strategy, LoadBalanceStrategy::Failover);
3340        } else {
3341            panic!("Expected LoadBalance step");
3342        }
3343    }
3344
3345    #[test]
3346    fn test_multicast_builder_all_config_setters() {
3347        let route = RouteBuilder::from("direct:start")
3348            .route_id("multicast-config-test")
3349            .multicast()
3350            .parallel(true)
3351            .parallel_limit(4)
3352            .stop_on_exception(true)
3353            .timeout(Duration::from_millis(300))
3354            .aggregation(MulticastStrategy::Original)
3355            .to("mock:a")
3356            .end_multicast()
3357            .build()
3358            .unwrap();
3359
3360        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3361            assert!(config.parallel);
3362            assert_eq!(config.parallel_limit, Some(4));
3363            assert!(config.stop_on_exception);
3364            assert_eq!(config.timeout, Some(Duration::from_millis(300)));
3365            assert!(matches!(config.aggregation, MulticastStrategy::Original));
3366        } else {
3367            panic!("Expected Multicast step");
3368        }
3369    }
3370
3371    #[test]
3372    fn test_build_canonical_rejects_unsupported_processor_step() {
3373        let err = RouteBuilder::from("direct:start")
3374            .route_id("canonical-reject")
3375            .set_header("k", Value::String("v".into()))
3376            .build_canonical()
3377            .unwrap_err();
3378
3379        assert!(format!("{err}").contains("does not support step `processor`"));
3380    }
3381
3382    // ── LoadBalance strategy-specific tests ─────────────────────────────────────
3383
3384    #[test]
3385    fn test_load_balance_builder_weighted_strategy() {
3386        let route = RouteBuilder::from("direct:start")
3387            .route_id("lb-weighted")
3388            .load_balance()
3389            .weighted(vec![
3390                ("direct:a".to_string(), 5),
3391                ("direct:b".to_string(), 2),
3392                ("direct:c".to_string(), 1),
3393            ])
3394            .to("mock:result")
3395            .end_load_balance()
3396            .build()
3397            .unwrap();
3398
3399        if let BuilderStep::LoadBalance { config, .. } = &route.steps()[0] {
3400            assert!(matches!(config.strategy, LoadBalanceStrategy::Weighted(_)));
3401        } else {
3402            panic!("Expected LoadBalance step");
3403        }
3404    }
3405
3406    #[test]
3407    fn test_load_balance_builder_failover_strategy() {
3408        let route = RouteBuilder::from("direct:start")
3409            .route_id("lb-failover")
3410            .load_balance()
3411            .failover()
3412            .to("mock:primary")
3413            .end_load_balance()
3414            .build()
3415            .unwrap();
3416
3417        if let BuilderStep::LoadBalance { config, .. } = &route.steps()[0] {
3418            assert_eq!(config.strategy, LoadBalanceStrategy::Failover);
3419        } else {
3420            panic!("Expected LoadBalance step");
3421        }
3422    }
3423
3424    // ── FilterInSplitBuilder tests ──────────────────────────────────────────────
3425
3426    #[test]
3427    fn test_filter_in_split_builder_typestate() {
3428        use camel_api::splitter::{SplitterConfig, split_body_lines};
3429
3430        let definition = RouteBuilder::from("timer:test")
3431            .route_id("filter-in-split")
3432            .split(SplitterConfig::new(split_body_lines()))
3433            .filter(|_ex| true)
3434            .to("mock:filtered")
3435            .end_filter()
3436            .end_split()
3437            .build()
3438            .unwrap();
3439
3440        assert_eq!(definition.steps().len(), 1);
3441        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
3442            assert_eq!(steps.len(), 1);
3443            assert!(matches!(&steps[0], BuilderStep::Filter { .. }));
3444        } else {
3445            panic!("Expected Split step");
3446        }
3447    }
3448
3449    #[test]
3450    fn test_filter_in_split_builder_multiple_steps() {
3451        use camel_api::splitter::{SplitterConfig, split_body_lines};
3452
3453        let definition = RouteBuilder::from("timer:test")
3454            .route_id("filter-in-split-multi")
3455            .split(SplitterConfig::new(split_body_lines()))
3456            .to("mock:before-filter")
3457            .filter(|_ex| true)
3458            .to("mock:inside-filter")
3459            .end_filter()
3460            .to("mock:after-filter")
3461            .end_split()
3462            .build()
3463            .unwrap();
3464
3465        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
3466            // To("before-filter") + Filter{...} + To("after-filter") = 3
3467            assert_eq!(steps.len(), 3);
3468        } else {
3469            panic!("Expected Split step");
3470        }
3471    }
3472
3473    // ── build_canonical tests ───────────────────────────────────────────────────
3474
3475    #[test]
3476    fn test_build_canonical_with_circuit_breaker() {
3477        use camel_api::circuit_breaker::CircuitBreakerConfig;
3478
3479        let spec = RouteBuilder::from("direct:start")
3480            .route_id("canonical-cb")
3481            .circuit_breaker(CircuitBreakerConfig::new().failure_threshold(10))
3482            .to("mock:result")
3483            .build_canonical()
3484            .unwrap();
3485
3486        let cb = spec.circuit_breaker.expect("circuit breaker should be set");
3487        assert_eq!(cb.failure_threshold, 10);
3488    }
3489
3490    #[test]
3491    fn test_build_canonical_rejects_custom_split_aggregation() {
3492        use camel_api::splitter::{SplitterConfig, split_body_lines};
3493
3494        let err = RouteBuilder::from("direct:start")
3495            .route_id("canonical-custom-split")
3496            .split(SplitterConfig::new(split_body_lines()).aggregation(
3497                camel_api::splitter::AggregationStrategy::Custom(Arc::new(|_, ex| ex)),
3498            ))
3499            .to("mock:frag")
3500            .end_split()
3501            .build_canonical()
3502            .unwrap_err();
3503
3504        // Split with closure-based expression is rejected in canonical v2.
3505        assert!(format!("{err}").contains("canonical v2 does not support step `split`"));
3506    }
3507
3508    #[test]
3509    fn test_build_canonical_rejects_custom_aggregate_strategy() {
3510        let err = RouteBuilder::from("direct:start")
3511            .route_id("canonical-custom-agg")
3512            .aggregate(
3513                AggregatorConfig::correlate_by("key")
3514                    .complete_when_size(2)
3515                    .strategy(AggregationStrategy::Custom(Arc::new(|_, ex| ex)))
3516                    .build()
3517                    .unwrap(),
3518            )
3519            .build_canonical()
3520            .unwrap_err();
3521
3522        assert!(format!("{err}").contains("custom aggregate strategy"));
3523    }
3524
3525    #[test]
3526    fn test_build_canonical_rejects_fn_correlation_strategy() {
3527        let err = RouteBuilder::from("direct:start")
3528            .route_id("canonical-fn-corr")
3529            .aggregate(AggregatorConfig {
3530                header_name: "key".to_string(),
3531                completion: CompletionMode::Single(CompletionCondition::Size(1)),
3532                correlation: CorrelationStrategy::Fn(Arc::new(|_| Some("key".to_string()))),
3533                strategy: AggregationStrategy::CollectAll,
3534                max_buckets: None,
3535                max_bucket_size: None,
3536                bucket_ttl: None,
3537                force_completion_on_stop: false,
3538                discard_on_timeout: false,
3539                max_timeout_tasks: 1024,
3540            })
3541            .build_canonical()
3542            .unwrap_err();
3543
3544        assert!(format!("{err}").contains("Fn correlation strategy"));
3545    }
3546
3547    #[test]
3548    fn test_build_canonical_rejects_predicate_completion() {
3549        let err = RouteBuilder::from("direct:start")
3550            .route_id("canonical-pred-completion")
3551            .aggregate(AggregatorConfig {
3552                header_name: "key".to_string(),
3553                completion: CompletionMode::Single(CompletionCondition::Predicate(Arc::new(
3554                    |_| false,
3555                ))),
3556                correlation: CorrelationStrategy::HeaderName("key".to_string()),
3557                strategy: AggregationStrategy::CollectAll,
3558                max_buckets: None,
3559                max_bucket_size: None,
3560                bucket_ttl: None,
3561                force_completion_on_stop: false,
3562                discard_on_timeout: false,
3563                max_timeout_tasks: 1024,
3564            })
3565            .build_canonical()
3566            .unwrap_err();
3567
3568        assert!(
3569            format!("{err}").contains("cannot reverse-map"),
3570            "reject message must explain forward-only: {}",
3571            err
3572        );
3573    }
3574
3575    #[test]
3576    fn extract_completion_fields_rejects_predicate_expr() {
3577        let mode = CompletionMode::Single(CompletionCondition::PredicateExpr {
3578            expr: "${body} == 'DONE'".to_string(),
3579            language: "simple".to_string(),
3580        });
3581        let result = extract_completion_fields(&mode);
3582        assert!(
3583            result.is_err(),
3584            "PredicateExpr must be rejected (forward-only)"
3585        );
3586        let msg = format!("{}", result.unwrap_err());
3587        assert!(
3588            msg.contains("cannot reverse-map"),
3589            "reject message must explain forward-only: {}",
3590            msg
3591        );
3592    }
3593
3594    #[test]
3595    fn extract_completion_fields_rejects_predicate_expr_any_mode() {
3596        let mode = CompletionMode::Any(vec![
3597            CompletionCondition::Size(5),
3598            CompletionCondition::PredicateExpr {
3599                expr: "${body} == 'DONE'".to_string(),
3600                language: "simple".to_string(),
3601            },
3602        ]);
3603        let result = extract_completion_fields(&mode);
3604        assert!(
3605            result.is_err(),
3606            "PredicateExpr in Any must be rejected (forward-only)"
3607        );
3608        let msg = format!("{}", result.unwrap_err());
3609        assert!(
3610            msg.contains("cannot reverse-map"),
3611            "reject message must explain forward-only: {}",
3612            msg
3613        );
3614    }
3615
3616    #[test]
3617    fn test_build_canonical_with_expression_correlation() {
3618        let spec = RouteBuilder::from("direct:start")
3619            .route_id("canonical-expr-corr")
3620            .aggregate(AggregatorConfig {
3621                header_name: "key".to_string(),
3622                completion: CompletionMode::Single(CompletionCondition::Size(1)),
3623                correlation: CorrelationStrategy::Expression {
3624                    expr: "header.key".to_string(),
3625                    language: "simple".to_string(),
3626                },
3627                strategy: AggregationStrategy::CollectAll,
3628                max_buckets: None,
3629                max_bucket_size: None,
3630                bucket_ttl: None,
3631                force_completion_on_stop: false,
3632                discard_on_timeout: false,
3633                max_timeout_tasks: 1024,
3634            })
3635            .build_canonical()
3636            .unwrap();
3637
3638        assert!(spec.steps.iter().any(|s| matches!(s, CanonicalStepSpec::Aggregate(a) if a.correlation_key == Some("header.key".to_string()))));
3639    }
3640
3641    #[test]
3642    fn test_build_canonical_split_rejected_with_closure_expression() {
3643        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
3644
3645        // Builder-based split uses closure expressions, which are not serializable.
3646        let err = RouteBuilder::from("direct:start")
3647            .route_id("canonical-split-last")
3648            .split(
3649                SplitterConfig::new(split_body_lines()).aggregation(AggregationStrategy::LastWins),
3650            )
3651            .to("mock:frag")
3652            .end_split()
3653            .build_canonical()
3654            .unwrap_err();
3655
3656        assert!(format!("{err}").contains("canonical v2 does not support step `split`"));
3657    }
3658
3659    // ── OnExceptionBuilder full chain tests ─────────────────────────────────────
3660
3661    #[test]
3662    fn test_on_exception_full_chain_retry_backoff_jitter_handled_by() {
3663        let definition = RouteBuilder::from("direct:start")
3664            .route_id("on-exception-full")
3665            .dead_letter_channel("log:dlc")
3666            .on_exception(|e| matches!(e, CamelError::Io(_)))
3667            .retry(5)
3668            .with_backoff(Duration::from_millis(10), 2.0, Duration::from_millis(500))
3669            .with_jitter(0.3)
3670            .handled_by("log:io-handler")
3671            .end_on_exception()
3672            .to("mock:out")
3673            .build()
3674            .unwrap();
3675
3676        let cfg = definition
3677            .error_handler_config()
3678            .expect("error handler should be set");
3679        assert_eq!(cfg.policies.len(), 1);
3680        let policy = &cfg.policies[0];
3681        let retry = policy.retry.as_ref().expect("retry should be set");
3682        assert_eq!(retry.max_attempts, 5);
3683        assert_eq!(retry.initial_delay, Duration::from_millis(10));
3684        assert_eq!(retry.multiplier, 2.0);
3685        assert_eq!(retry.max_delay, Duration::from_millis(500));
3686        assert!((retry.jitter_factor - 0.3).abs() < f64::EPSILON);
3687        assert_eq!(policy.handled_by.as_deref(), Some("log:io-handler"));
3688    }
3689
3690    #[test]
3691    fn test_on_exception_jitter_clamped_to_valid_range() {
3692        let definition = RouteBuilder::from("direct:start")
3693            .route_id("jitter-clamp")
3694            .on_exception(|_e| true)
3695            .retry(1)
3696            .with_jitter(5.0)
3697            .end_on_exception()
3698            .to("mock:out")
3699            .build()
3700            .unwrap();
3701
3702        let cfg = definition.error_handler_config().unwrap();
3703        let retry = cfg.policies[0].retry.as_ref().unwrap();
3704        assert!((retry.jitter_factor - 1.0).abs() < f64::EPSILON);
3705    }
3706
3707    // ── StepAccumulator: process_fn, convert_body_to, bean ──────────────────────
3708
3709    #[test]
3710    fn test_builder_process_fn_adds_processor_step() {
3711        use camel_api::BoxProcessorExt;
3712        let processor = BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }));
3713        let definition = RouteBuilder::from("timer:tick")
3714            .route_id("process-fn-test")
3715            .process_fn(processor)
3716            .build()
3717            .unwrap();
3718
3719        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3720    }
3721
3722    #[test]
3723    fn test_builder_convert_body_to_adds_processor_step() {
3724        let definition = RouteBuilder::from("timer:tick")
3725            .route_id("convert-body-test")
3726            .convert_body_to(BodyType::Json)
3727            .build()
3728            .unwrap();
3729
3730        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3731    }
3732
3733    #[test]
3734    fn test_builder_bean_adds_bean_step() {
3735        let definition = RouteBuilder::from("timer:tick")
3736            .route_id("bean-test")
3737            .bean("myBean", "process")
3738            .build()
3739            .unwrap();
3740
3741        assert!(
3742            matches!(&definition.steps()[0], BuilderStep::Bean { name, method }
3743            if name == "myBean" && method == "process")
3744        );
3745    }
3746
3747    // ── Throttle strategy-specific tests ────────────────────────────────────────
3748
3749    #[test]
3750    fn test_throttle_builder_delay_strategy() {
3751        let definition = RouteBuilder::from("timer:tick")
3752            .route_id("throttle-delay")
3753            .throttle(10, Duration::from_secs(1))
3754            .strategy(ThrottleStrategy::Delay)
3755            .to("mock:result")
3756            .end_throttle()
3757            .build()
3758            .unwrap();
3759
3760        if let BuilderStep::Throttle { config, .. } = &definition.steps()[0] {
3761            assert_eq!(config.strategy, ThrottleStrategy::Delay);
3762        } else {
3763            panic!("Expected Throttle step");
3764        }
3765    }
3766
3767    #[test]
3768    fn test_throttle_builder_drop_strategy() {
3769        let definition = RouteBuilder::from("timer:tick")
3770            .route_id("throttle-drop")
3771            .throttle(10, Duration::from_secs(1))
3772            .strategy(ThrottleStrategy::Drop)
3773            .to("mock:result")
3774            .end_throttle()
3775            .build()
3776            .unwrap();
3777
3778        if let BuilderStep::Throttle { config, .. } = &definition.steps()[0] {
3779            assert_eq!(config.strategy, ThrottleStrategy::Drop);
3780        } else {
3781            panic!("Expected Throttle step");
3782        }
3783    }
3784
3785    // ── LoopInLoopBuilder with loop_while ───────────────────────────────────────
3786
3787    #[test]
3788    fn test_nested_loop_while_builder() {
3789        use camel_api::loop_eip::LoopMode;
3790
3791        let def = RouteBuilder::from("direct:start")
3792            .route_id("nested-loop-while")
3793            .loop_count(2)
3794            .to("mock:outer")
3795            .loop_while(|_ex| true)
3796            .to("mock:inner")
3797            .end_loop()
3798            .end_loop()
3799            .build()
3800            .unwrap();
3801
3802        assert_eq!(def.steps().len(), 1);
3803        if let BuilderStep::Loop { steps, .. } = &def.steps()[0] {
3804            assert_eq!(steps.len(), 2);
3805            if let BuilderStep::Loop { config, .. } = &steps[1] {
3806                assert!(matches!(config.mode, LoopMode::While(_)));
3807            } else {
3808                panic!("Expected inner Loop step");
3809            }
3810        } else {
3811            panic!("Expected outer Loop step");
3812        }
3813    }
3814
3815    // ── Choice with multiple whens + otherwise ──────────────────────────────────
3816
3817    #[test]
3818    fn test_choice_builder_multiple_whens_with_otherwise() {
3819        let definition = RouteBuilder::from("timer:tick")
3820            .route_id("choice-multi-otherwise")
3821            .choice()
3822            .when(|ex: &Exchange| ex.input.header("a").is_some())
3823            .to("mock:a")
3824            .end_when()
3825            .when(|ex: &Exchange| ex.input.header("b").is_some())
3826            .to("mock:b")
3827            .end_when()
3828            .when(|ex: &Exchange| ex.input.header("c").is_some())
3829            .to("mock:c")
3830            .end_when()
3831            .otherwise()
3832            .to("mock:fallback")
3833            .end_otherwise()
3834            .end_choice()
3835            .build()
3836            .unwrap();
3837
3838        if let BuilderStep::Choice { whens, otherwise } = &definition.steps()[0] {
3839            assert_eq!(whens.len(), 3);
3840            assert!(otherwise.is_some());
3841            assert_eq!(otherwise.as_ref().unwrap().len(), 1);
3842        } else {
3843            panic!("Expected Choice step");
3844        }
3845    }
3846
3847    // ── Multicast individual config tests ───────────────────────────────────────
3848
3849    #[test]
3850    fn test_multicast_builder_parallel_only() {
3851        let route = RouteBuilder::from("direct:start")
3852            .route_id("multicast-parallel")
3853            .multicast()
3854            .parallel(true)
3855            .to("mock:a")
3856            .end_multicast()
3857            .build()
3858            .unwrap();
3859
3860        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3861            assert!(config.parallel);
3862            assert_eq!(config.parallel_limit, None);
3863        } else {
3864            panic!("Expected Multicast step");
3865        }
3866    }
3867
3868    #[test]
3869    fn test_multicast_builder_timeout_only() {
3870        let route = RouteBuilder::from("direct:start")
3871            .route_id("multicast-timeout")
3872            .multicast()
3873            .timeout(Duration::from_secs(5))
3874            .to("mock:a")
3875            .end_multicast()
3876            .build()
3877            .unwrap();
3878
3879        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3880            assert_eq!(config.timeout, Some(Duration::from_secs(5)));
3881        } else {
3882            panic!("Expected Multicast step");
3883        }
3884    }
3885
3886    #[test]
3887    fn test_multicast_builder_aggregation_collect_all() {
3888        let route = RouteBuilder::from("direct:start")
3889            .route_id("multicast-collect")
3890            .multicast()
3891            .aggregation(MulticastStrategy::CollectAll)
3892            .to("mock:a")
3893            .end_multicast()
3894            .build()
3895            .unwrap();
3896
3897        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3898            assert!(matches!(config.aggregation, MulticastStrategy::CollectAll));
3899        } else {
3900            panic!("Expected Multicast step");
3901        }
3902    }
3903
3904    // ── extract_completion_fields: Any mode with multiple conditions ────────────
3905
3906    #[test]
3907    fn test_build_canonical_aggregate_any_completion_mode() {
3908        let spec = RouteBuilder::from("direct:start")
3909            .route_id("canonical-any-completion")
3910            .aggregate(
3911                AggregatorConfig::correlate_by("key")
3912                    .complete_on_size_or_timeout(10, Duration::from_secs(30))
3913                    .build()
3914                    .unwrap(),
3915            )
3916            .build_canonical()
3917            .unwrap();
3918
3919        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3920            assert_eq!(agg.completion_size, Some(10));
3921            assert_eq!(agg.completion_timeout_ms, Some(30_000));
3922        } else {
3923            panic!("Expected Aggregate step");
3924        }
3925    }
3926
3927    #[test]
3928    fn test_build_canonical_aggregate_timeout_completion() {
3929        let spec = RouteBuilder::from("direct:start")
3930            .route_id("canonical-timeout-completion")
3931            .aggregate(
3932                AggregatorConfig::correlate_by("key")
3933                    .complete_on_timeout(Duration::from_millis(500))
3934                    .build()
3935                    .unwrap(),
3936            )
3937            .build_canonical()
3938            .unwrap();
3939
3940        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3941            assert_eq!(agg.completion_size, None);
3942            assert_eq!(agg.completion_timeout_ms, Some(500));
3943        } else {
3944            panic!("Expected Aggregate step");
3945        }
3946    }
3947
3948    // ── canonicalize_aggregate: discard_on_timeout and force_completion_on_stop ─
3949
3950    #[test]
3951    fn test_build_canonical_aggregate_discard_on_timeout() {
3952        use camel_api::aggregator::AggregatorConfig;
3953
3954        let spec = RouteBuilder::from("direct:start")
3955            .route_id("canonical-discard-timeout")
3956            .aggregate(
3957                AggregatorConfig::correlate_by("key")
3958                    .complete_when_size(1)
3959                    .discard_on_timeout(true)
3960                    .build()
3961                    .unwrap(),
3962            )
3963            .build_canonical()
3964            .unwrap();
3965
3966        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3967            assert_eq!(agg.discard_on_timeout, Some(true));
3968        } else {
3969            panic!("Expected Aggregate step");
3970        }
3971    }
3972
3973    #[test]
3974    fn test_build_canonical_aggregate_force_completion_on_stop() {
3975        use camel_api::aggregator::AggregatorConfig;
3976
3977        let spec = RouteBuilder::from("direct:start")
3978            .route_id("canonical-force-stop")
3979            .aggregate(
3980                AggregatorConfig::correlate_by("key")
3981                    .complete_when_size(1)
3982                    .force_completion_on_stop(true)
3983                    .build()
3984                    .unwrap(),
3985            )
3986            .build_canonical()
3987            .unwrap();
3988
3989        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3990            assert_eq!(agg.force_completion_on_stop, Some(true));
3991        } else {
3992            panic!("Expected Aggregate step");
3993        }
3994    }
3995
3996    // ── build_canonical: max_buckets and bucket_ttl ─────────────────────────────
3997
3998    #[test]
3999    fn test_build_canonical_aggregate_max_buckets_and_ttl() {
4000        use camel_api::aggregator::AggregatorConfig;
4001
4002        let spec = RouteBuilder::from("direct:start")
4003            .route_id("canonical-buckets-ttl")
4004            .aggregate(
4005                AggregatorConfig::correlate_by("key")
4006                    .complete_when_size(1)
4007                    .max_buckets(100)
4008                    .bucket_ttl(Duration::from_secs(60))
4009                    .build()
4010                    .unwrap(),
4011            )
4012            .build_canonical()
4013            .unwrap();
4014
4015        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
4016            assert_eq!(agg.max_buckets, Some(100));
4017            assert_eq!(agg.bucket_ttl_ms, Some(60_000));
4018        } else {
4019            panic!("Expected Aggregate step");
4020        }
4021    }
4022
4023    // ── SplitBuilder with filter inside ─────────────────────────────────────────
4024
4025    #[test]
4026    fn test_split_builder_with_filter_inside() {
4027        use camel_api::splitter::{SplitterConfig, split_body_lines};
4028
4029        let definition = RouteBuilder::from("timer:test")
4030            .route_id("split-with-filter")
4031            .split(SplitterConfig::new(split_body_lines()))
4032            .filter(|_ex| true)
4033            .to("mock:filtered-frag")
4034            .end_filter()
4035            .end_split()
4036            .build()
4037            .unwrap();
4038
4039        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
4040            assert_eq!(steps.len(), 1);
4041            assert!(matches!(&steps[0], BuilderStep::Filter { .. }));
4042        } else {
4043            panic!("Expected Split step");
4044        }
4045    }
4046
4047    // ── WireTap additional tests ────────────────────────────────────────────────
4048
4049    #[test]
4050    fn test_wire_tap_multiple_taps() {
4051        let definition = RouteBuilder::from("timer:tick")
4052            .route_id("multi-wire-tap")
4053            .wire_tap("mock:tap1")
4054            .wire_tap("mock:tap2")
4055            .to("mock:result")
4056            .build()
4057            .unwrap();
4058
4059        assert_eq!(definition.steps().len(), 3);
4060        assert!(
4061            matches!(&definition.steps()[0], BuilderStep::WireTap { uri } if uri == "mock:tap1")
4062        );
4063        assert!(
4064            matches!(&definition.steps()[1], BuilderStep::WireTap { uri } if uri == "mock:tap2")
4065        );
4066    }
4067
4068    // ── Error handler: explicit config after shorthand → Mixed mode ─────────────
4069
4070    #[test]
4071    fn test_builder_shorthand_then_explicit_mixed_mode() {
4072        let result = RouteBuilder::from("direct:start")
4073            .route_id("mixed-mode-2")
4074            .dead_letter_channel("log:dlc")
4075            .error_handler(ErrorHandlerConfig::log_only())
4076            .to("mock:out")
4077            .build();
4078
4079        let err = result.err().expect("mixed mode should fail");
4080        assert!(format!("{err}").contains("mixed error handler modes"));
4081    }
4082
4083    // ── build_canonical: empty from_uri error ───────────────────────────────────
4084
4085    #[test]
4086    fn test_build_canonical_empty_from_uri_errors() {
4087        let result = RouteBuilder::from("").route_id("test").build_canonical();
4088        assert!(result.is_err());
4089    }
4090
4091    #[test]
4092    fn test_build_canonical_missing_route_id_errors() {
4093        let result = RouteBuilder::from("direct:start").build_canonical();
4094        assert!(result.is_err());
4095        let err = result.unwrap_err().to_string();
4096        assert!(err.contains("route_id"));
4097    }
4098
4099    // ── SplitBuilder: aggregate inside split ────────────────────────────────────
4100
4101    #[test]
4102    fn test_split_builder_with_aggregate_inside() {
4103        use camel_api::aggregator::AggregatorConfig;
4104        use camel_api::splitter::{SplitterConfig, split_body_lines};
4105
4106        let definition = RouteBuilder::from("timer:test")
4107            .route_id("split-agg")
4108            .split(SplitterConfig::new(split_body_lines()))
4109            .aggregate(
4110                AggregatorConfig::correlate_by("frag-key")
4111                    .complete_when_size(3)
4112                    .build()
4113                    .unwrap(),
4114            )
4115            .end_split()
4116            .build()
4117            .unwrap();
4118
4119        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
4120            assert_eq!(steps.len(), 1);
4121            assert!(matches!(&steps[0], BuilderStep::Aggregate { .. }));
4122        } else {
4123            panic!("Expected Split step");
4124        }
4125    }
4126
4127    // ── Throttle: steps collected inside throttle scope ─────────────────────────
4128
4129    #[test]
4130    fn test_throttle_builder_with_steps_inside() {
4131        let definition = RouteBuilder::from("timer:tick")
4132            .route_id("throttle-steps")
4133            .throttle(10, Duration::from_secs(1))
4134            .set_header("throttled", Value::Bool(true))
4135            .to("mock:throttled")
4136            .end_throttle()
4137            .build()
4138            .unwrap();
4139
4140        if let BuilderStep::Throttle { steps, .. } = &definition.steps()[0] {
4141            assert_eq!(steps.len(), 2);
4142        } else {
4143            panic!("Expected Throttle step");
4144        }
4145    }
4146
4147    // ── LoadBalance: steps collected inside scope ───────────────────────────────
4148
4149    #[test]
4150    fn test_load_balance_builder_with_steps_inside() {
4151        let definition = RouteBuilder::from("timer:tick")
4152            .route_id("lb-steps")
4153            .load_balance()
4154            .round_robin()
4155            .set_header("lb", Value::Bool(true))
4156            .to("mock:lb")
4157            .end_load_balance()
4158            .build()
4159            .unwrap();
4160
4161        if let BuilderStep::LoadBalance { steps, .. } = &definition.steps()[0] {
4162            assert_eq!(steps.len(), 2);
4163        } else {
4164            panic!("Expected LoadBalance step");
4165        }
4166    }
4167
4168    // ── Multicast: steps collected inside scope ─────────────────────────────────
4169
4170    #[test]
4171    fn test_multicast_builder_with_steps_inside() {
4172        let definition = RouteBuilder::from("timer:tick")
4173            .route_id("multicast-steps")
4174            .multicast()
4175            .set_header("mc", Value::Bool(true))
4176            .to("mock:multicast")
4177            .end_multicast()
4178            .build()
4179            .unwrap();
4180
4181        if let BuilderStep::Multicast { steps, .. } = &definition.steps()[0] {
4182            assert_eq!(steps.len(), 2);
4183        } else {
4184            panic!("Expected Multicast step");
4185        }
4186    }
4187
4188    // ── LoopBuilder: steps collected inside loop scope ──────────────────────────
4189
4190    #[test]
4191    fn test_loop_builder_with_steps_inside() {
4192        let definition = RouteBuilder::from("timer:tick")
4193            .route_id("loop-steps")
4194            .loop_count(3)
4195            .set_header("loop", Value::Bool(true))
4196            .to("mock:loop")
4197            .end_loop()
4198            .build()
4199            .unwrap();
4200
4201        if let BuilderStep::Loop { steps, .. } = &definition.steps()[0] {
4202            assert_eq!(steps.len(), 2);
4203        } else {
4204            panic!("Expected Loop step");
4205        }
4206    }
4207
4208    // ── canonical_step_name coverage for remaining variants ─────────────────────
4209
4210    #[test]
4211    fn test_build_canonical_rejects_loop_step() {
4212        let err = RouteBuilder::from("direct:start")
4213            .route_id("canonical-loop")
4214            .loop_count(3)
4215            .to("mock:loop")
4216            .end_loop()
4217            .build_canonical()
4218            .unwrap_err();
4219
4220        assert!(format!("{err}").contains("does not support step `loop`"));
4221    }
4222
4223    #[test]
4224    fn test_build_canonical_rejects_multicast_step() {
4225        let err = RouteBuilder::from("direct:start")
4226            .route_id("canonical-multicast")
4227            .multicast()
4228            .to("mock:a")
4229            .end_multicast()
4230            .build_canonical()
4231            .unwrap_err();
4232
4233        assert!(format!("{err}").contains("does not support step `multicast`"));
4234    }
4235
4236    #[test]
4237    fn test_build_canonical_rejects_throttle_step() {
4238        let err = RouteBuilder::from("direct:start")
4239            .route_id("canonical-throttle")
4240            .throttle(10, Duration::from_secs(1))
4241            .to("mock:result")
4242            .end_throttle()
4243            .build_canonical()
4244            .unwrap_err();
4245
4246        assert!(format!("{err}").contains("does not support step `throttle`"));
4247    }
4248
4249    #[test]
4250    fn test_build_canonical_rejects_load_balancer_step() {
4251        let err = RouteBuilder::from("direct:start")
4252            .route_id("canonical-lb")
4253            .load_balance()
4254            .round_robin()
4255            .to("mock:result")
4256            .end_load_balance()
4257            .build_canonical()
4258            .unwrap_err();
4259
4260        assert!(format!("{err}").contains("does not support step `load_balancer`"));
4261    }
4262
4263    #[test]
4264    fn test_build_canonical_rejects_bean_step() {
4265        let err = RouteBuilder::from("direct:start")
4266            .route_id("canonical-bean")
4267            .bean("myBean", "process")
4268            .build_canonical()
4269            .unwrap_err();
4270
4271        assert!(format!("{err}").contains("does not support step `bean`"));
4272    }
4273
4274    #[test]
4275    fn test_build_canonical_rejects_script_step() {
4276        let err = RouteBuilder::from("direct:start")
4277            .route_id("canonical-script")
4278            .script("rhai", "x = 1")
4279            .build_canonical()
4280            .unwrap_err();
4281
4282        assert!(format!("{err}").contains("does not support step `script`"));
4283    }
4284
4285    #[test]
4286    fn test_build_canonical_accepts_delay_step() {
4287        let spec = RouteBuilder::from("direct:start")
4288            .route_id("canonical-delay")
4289            .delay(Duration::from_millis(100))
4290            .build_canonical()
4291            .unwrap();
4292
4293        assert!(
4294            spec.steps.iter().any(
4295                |s| matches!(s, CanonicalStepSpec::Delay { delay_ms, .. } if *delay_ms == 100)
4296            )
4297        );
4298    }
4299
4300    #[test]
4301    fn test_build_canonical_accepts_wire_tap_step() {
4302        let spec = RouteBuilder::from("direct:start")
4303            .route_id("canonical-wiretap")
4304            .wire_tap("mock:tap")
4305            .build_canonical()
4306            .unwrap();
4307
4308        assert!(
4309            spec.steps
4310                .iter()
4311                .any(|s| matches!(s, CanonicalStepSpec::WireTap { uri } if uri == "mock:tap"))
4312        );
4313    }
4314
4315    #[test]
4316    fn test_build_canonical_rejects_dynamic_router_step() {
4317        let err = RouteBuilder::from("direct:start")
4318            .route_id("canonical-dyn-router")
4319            .dynamic_router(Arc::new(|_| Some("mock:a".to_string())))
4320            .build_canonical()
4321            .unwrap_err();
4322
4323        assert!(format!("{err}").contains("does not support step `dynamic_router`"));
4324    }
4325
4326    #[test]
4327    fn test_build_canonical_rejects_routing_slip_step() {
4328        let err = RouteBuilder::from("direct:start")
4329            .route_id("canonical-routing-slip")
4330            .routing_slip(Arc::new(|_| Some("mock:a".to_string())))
4331            .build_canonical()
4332            .unwrap_err();
4333
4334        assert!(format!("{err}").contains("does not support step `routing_slip`"));
4335    }
4336
4337    #[test]
4338    fn test_build_canonical_rejects_recipient_list_step() {
4339        let err = RouteBuilder::from("direct:start")
4340            .route_id("canonical-recipient")
4341            .recipient_list(Arc::new(|_| "mock:a".to_string()))
4342            .build_canonical()
4343            .unwrap_err();
4344
4345        assert!(format!("{err}").contains("does not support step `recipient_list`"));
4346    }
4347
4348    // ── extract_completion_fields: Any mode with predicate → error ──────────────
4349
4350    #[test]
4351    fn test_build_canonical_rejects_any_mode_with_predicate() {
4352        let err = RouteBuilder::from("direct:start")
4353            .route_id("canonical-any-pred")
4354            .aggregate(AggregatorConfig {
4355                header_name: "key".to_string(),
4356                completion: CompletionMode::Any(vec![
4357                    CompletionCondition::Size(5),
4358                    CompletionCondition::Predicate(Arc::new(|_| false)),
4359                ]),
4360                correlation: CorrelationStrategy::HeaderName("key".to_string()),
4361                strategy: AggregationStrategy::CollectAll,
4362                max_buckets: None,
4363                max_bucket_size: None,
4364                bucket_ttl: None,
4365                force_completion_on_stop: false,
4366                discard_on_timeout: false,
4367                max_timeout_tasks: 1024,
4368            })
4369            .build_canonical()
4370            .unwrap_err();
4371
4372        assert!(
4373            format!("{err}").contains("cannot reverse-map"),
4374            "reject message must explain forward-only: {}",
4375            err
4376        );
4377    }
4378
4379    // ── BUILDER-004: Validation errors for missing required fields ────────────
4380
4381    #[test]
4382    fn test_builder_validation_missing_from_uri() {
4383        let result = RouteBuilder::from("")
4384            .route_id("missing-uri-route")
4385            .to("log:info")
4386            .build();
4387        assert!(result.is_err(), "empty from URI should fail validation");
4388        let err = result.err().unwrap().to_string();
4389        assert!(
4390            err.contains("'from'") || err.contains("URI"),
4391            "error should mention from/URI, got: {err}"
4392        );
4393    }
4394
4395    #[test]
4396    fn test_builder_validation_invalid_step_uri_scheme() {
4397        let result = RouteBuilder::from("timer:tick")
4398            .route_id("bad-step-route")
4399            .to("not-a-valid-uri") // no scheme
4400            .build();
4401        // The builder itself accepts any URI string; validation happens at
4402        // resolution time. Verify the build succeeds (step URI is deferred).
4403        assert!(
4404            result.is_ok(),
4405            "builder should accept opaque step URIs; resolution happens later"
4406        );
4407    }
4408
4409    // ── rc-p9vq: Duplicate route IDs ──────────────────────────────────────
4410
4411    #[test]
4412    fn test_builder_duplicate_route_ids_produce_identical_definitions() {
4413        // The builder itself doesn't check for duplicates (that's context-level).
4414        // Verify both builds succeed with the same ID — detection is TODO(rc-p9vq).
4415        let route1 = RouteBuilder::from("direct:a")
4416            .route_id("dup-route")
4417            .to("mock:out")
4418            .build();
4419        let route2 = RouteBuilder::from("direct:b")
4420            .route_id("dup-route")
4421            .to("mock:out")
4422            .build();
4423
4424        assert!(route1.is_ok());
4425        assert!(route2.is_ok());
4426        assert_eq!(route1.unwrap().route_id(), route2.unwrap().route_id());
4427    }
4428
4429    #[test]
4430    fn test_builder_clone_reuse_as_template() {
4431        // rc-8m5o: a partially-built RouteBuilder can be cloned and reused as a
4432        // template, then each clone varied independently before build().
4433        let template = RouteBuilder::from("direct:in")
4434            .set_header("stage", Value::String("shared".into()))
4435            .log("shared prefix", LogLevel::Info);
4436
4437        let route_a = template
4438            .clone()
4439            .route_id("route-a")
4440            .to("mock:a")
4441            .build()
4442            .expect("clone A builds");
4443        let route_b = template
4444            .route_id("route-b")
4445            .to("mock:b")
4446            .build()
4447            .expect("clone B builds");
4448
4449        assert_eq!(route_a.route_id(), "route-a");
4450        assert_eq!(route_b.route_id(), "route-b");
4451        // Shared template steps (set_header + log) are present in both, plus the
4452        // per-clone `to` step: the clone is a deep copy, not an alias.
4453        assert_eq!(route_a.steps().len(), route_b.steps().len());
4454        assert_eq!(route_a.from_uri(), route_b.from_uri());
4455    }
4456}
camel_builder/lib.rs

camel_builder/
lib.rs
