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        bucket_ttl_ms,
1256        completion_predicate: None,
1257    })
1258}
1259
1260fn canonicalize_circuit_breaker(
1261    config: CircuitBreakerConfig,
1262) -> Result<CanonicalCircuitBreakerSpec, CamelError> {
1263    if config.fallback.is_some() {
1264        return Err(CamelError::RouteError(
1265            "canonical v2 does not support circuit breaker `fallback` (opaque BoxProcessor \
1266             cannot reverse-map to canonical steps); build the canonical spec directly"
1267                .to_string(),
1268        ));
1269    }
1270    Ok(CanonicalCircuitBreakerSpec {
1271        failure_threshold: config.failure_threshold,
1272        open_duration_ms: u64::try_from(config.open_duration.as_millis()).unwrap_or(u64::MAX),
1273        fallback: Vec::new(),
1274    })
1275}
1276
1277fn canonical_step_name(step: &BuilderStep) -> &'static str {
1278    match step {
1279        BuilderStep::Processor(_) => "processor",
1280        BuilderStep::To(_) => "to",
1281        BuilderStep::Stop => "stop",
1282        BuilderStep::Log { .. } => "log",
1283        BuilderStep::DeclarativeSetHeader { .. } => "set_header",
1284        BuilderStep::DeclarativeSetHeaderIfAbsent { .. } => "set_header_if_absent",
1285        BuilderStep::DeclarativeRemoveHeader { .. } => "remove_header",
1286        BuilderStep::DeclarativeSetBody { .. } => "set_body",
1287        BuilderStep::DeclarativeFilter { .. } => "filter",
1288        BuilderStep::DeclarativeChoice { .. } => "choice",
1289        BuilderStep::DeclarativeScript { .. } => "script",
1290        BuilderStep::DeclarativeFunction { .. } => "function",
1291        BuilderStep::DeclarativeSplit { .. } => "split",
1292        BuilderStep::Split { .. } => "split",
1293        BuilderStep::Loop { .. } | BuilderStep::DeclarativeLoop { .. } => "loop",
1294        BuilderStep::Aggregate { .. } => "aggregate",
1295        BuilderStep::Filter { .. } => "filter",
1296        BuilderStep::Choice { .. } => "choice",
1297        BuilderStep::WireTap { .. } => "wire_tap",
1298        BuilderStep::Delay { .. } => "delay",
1299        BuilderStep::Multicast { .. } => "multicast",
1300        BuilderStep::DeclarativeLog { .. } => "log",
1301        BuilderStep::Bean { .. } => "bean",
1302        BuilderStep::Script { .. } => "script",
1303        BuilderStep::Throttle { .. } => "throttle",
1304        BuilderStep::LoadBalance { .. } => "load_balancer",
1305        BuilderStep::DynamicRouter { .. } => "dynamic_router",
1306        BuilderStep::RoutingSlip { .. } => "routing_slip",
1307        BuilderStep::DeclarativeDynamicRouter { .. } => "declarative_dynamic_router",
1308        BuilderStep::DeclarativeRoutingSlip { .. } => "declarative_routing_slip",
1309        BuilderStep::RecipientList { .. } => "recipient_list",
1310        BuilderStep::DeclarativeRecipientList { .. } => "declarative_recipient_list",
1311        BuilderStep::DeclarativeSetProperty { .. } => "set_property",
1312        BuilderStep::DeclarativeStreamSplit { .. } => "stream_split",
1313        BuilderStep::Enrich { .. } => "enrich",
1314        BuilderStep::PollEnrich { .. } => "poll_enrich",
1315        BuilderStep::Validate { .. } => "validate",
1316        BuilderStep::IdempotentConsumer { .. } => "idempotent_consumer",
1317        BuilderStep::ClaimCheck { .. } => "claim_check",
1318        BuilderStep::Cache { .. } => "cache",
1319        BuilderStep::CacheInvalidate { .. } => "cache_invalidate",
1320        BuilderStep::CacheClear { .. } => "cache_clear",
1321        BuilderStep::CacheStats { .. } => "cache_stats",
1322        BuilderStep::CachePeekStale { .. } => "cache_peek_stale",
1323        BuilderStep::Sampling { .. } => "sampling",
1324        BuilderStep::Sort { .. } => "sort",
1325        BuilderStep::DeclarativeDoTry { .. } => "do_try",
1326        BuilderStep::Resequence { .. } => "resequence",
1327    }
1328}
1329
1330impl StepAccumulator for RouteBuilder {
1331    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1332        &mut self.steps
1333    }
1334}
1335
1336/// Builder for the sub-pipeline within a `.split()` ... `.end_split()` block.
1337///
1338/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1339/// but NOT `.build()` and NOT `.split()` (no nested splits).
1340///
1341/// Calling `.end_split()` packages the sub-steps into a `BuilderStep::Split`
1342/// and returns the parent `RouteBuilder`.
1343pub struct SplitBuilder {
1344    parent: RouteBuilder,
1345    config: SplitterConfig,
1346    steps: Vec<BuilderStep>,
1347}
1348
1349impl SplitBuilder {
1350    /// Open a filter scope within the split sub-pipeline.
1351    pub fn filter<F>(self, predicate: F) -> FilterInSplitBuilder
1352    where
1353        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
1354    {
1355        FilterInSplitBuilder {
1356            parent: self,
1357            predicate: camel_api::FilterPredicate::new(predicate),
1358            steps: vec![],
1359        }
1360    }
1361
1362    /// Close the split scope. Packages the accumulated sub-steps into a
1363    /// `BuilderStep::Split` and returns the parent `RouteBuilder`.
1364    pub fn end_split(mut self) -> RouteBuilder {
1365        let split_step = BuilderStep::Split {
1366            config: self.config,
1367            steps: self.steps,
1368        };
1369        self.parent.steps.push(split_step);
1370        self.parent
1371    }
1372}
1373
1374impl StepAccumulator for SplitBuilder {
1375    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1376        &mut self.steps
1377    }
1378}
1379
1380/// Builder for the sub-pipeline within a `.filter()` ... `.end_filter()` block.
1381pub struct FilterBuilder {
1382    parent: RouteBuilder,
1383    predicate: FilterPredicate,
1384    steps: Vec<BuilderStep>,
1385}
1386
1387impl FilterBuilder {
1388    /// Close the filter scope. Packages the accumulated sub-steps into a
1389    /// `BuilderStep::Filter` and returns the parent `RouteBuilder`.
1390    pub fn end_filter(mut self) -> RouteBuilder {
1391        let step = BuilderStep::Filter {
1392            predicate: self.predicate,
1393            steps: self.steps,
1394        };
1395        self.parent.steps.push(step);
1396        self.parent
1397    }
1398}
1399
1400impl StepAccumulator for FilterBuilder {
1401    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1402        &mut self.steps
1403    }
1404}
1405
1406/// Builder for a filter scope nested inside a `.split()` block.
1407pub struct FilterInSplitBuilder {
1408    parent: SplitBuilder,
1409    predicate: FilterPredicate,
1410    steps: Vec<BuilderStep>,
1411}
1412
1413impl FilterInSplitBuilder {
1414    /// Close the filter scope and return the parent `SplitBuilder`.
1415    pub fn end_filter(mut self) -> SplitBuilder {
1416        let step = BuilderStep::Filter {
1417            predicate: self.predicate,
1418            steps: self.steps,
1419        };
1420        self.parent.steps.push(step);
1421        self.parent
1422    }
1423}
1424
1425impl StepAccumulator for FilterInSplitBuilder {
1426    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1427        &mut self.steps
1428    }
1429}
1430
1431// ── Choice/When/Otherwise builders ─────────────────────────────────────────
1432
1433/// Builder for a `.choice()` ... `.end_choice()` block.
1434///
1435/// Accumulates `when` clauses and an optional `otherwise` clause.
1436/// Cannot call `.build()` until `.end_choice()` is called.
1437pub struct ChoiceBuilder {
1438    parent: RouteBuilder,
1439    whens: Vec<WhenStep>,
1440    _otherwise: Option<Vec<BuilderStep>>,
1441}
1442
1443impl ChoiceBuilder {
1444    /// Open a `when` clause. Only exchanges matching `predicate` will be
1445    /// processed by the steps inside the `.when()` ... `.end_when()` scope.
1446    pub fn when<F>(self, predicate: F) -> WhenBuilder
1447    where
1448        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
1449    {
1450        WhenBuilder {
1451            parent: self,
1452            predicate: camel_api::FilterPredicate::new(predicate),
1453            steps: vec![],
1454        }
1455    }
1456
1457    /// Open an `otherwise` clause. Executed when no `when` predicate matched.
1458    ///
1459    /// Only one `otherwise` is allowed per `choice`. Call this after all `.when()` clauses.
1460    pub fn otherwise(self) -> OtherwiseBuilder {
1461        OtherwiseBuilder {
1462            parent: self,
1463            steps: vec![],
1464        }
1465    }
1466
1467    /// Close the choice scope. Packages all accumulated `when` clauses and
1468    /// optional `otherwise` into a `BuilderStep::Choice` and returns the
1469    /// parent `RouteBuilder`.
1470    pub fn end_choice(mut self) -> RouteBuilder {
1471        let step = BuilderStep::Choice {
1472            whens: self.whens,
1473            otherwise: self._otherwise,
1474        };
1475        self.parent.steps.push(step);
1476        self.parent
1477    }
1478}
1479
1480/// Builder for the sub-pipeline within a `.when()` ... `.end_when()` block.
1481pub struct WhenBuilder {
1482    parent: ChoiceBuilder,
1483    predicate: camel_api::FilterPredicate,
1484    steps: Vec<BuilderStep>,
1485}
1486
1487impl WhenBuilder {
1488    /// Close the when scope. Packages the accumulated sub-steps into a
1489    /// `WhenStep` and returns the parent `ChoiceBuilder`.
1490    pub fn end_when(mut self) -> ChoiceBuilder {
1491        self.parent.whens.push(WhenStep {
1492            predicate: self.predicate,
1493            steps: self.steps,
1494        });
1495        self.parent
1496    }
1497}
1498
1499impl StepAccumulator for WhenBuilder {
1500    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1501        &mut self.steps
1502    }
1503}
1504
1505/// Builder for the sub-pipeline within an `.otherwise()` ... `.end_otherwise()` block.
1506pub struct OtherwiseBuilder {
1507    parent: ChoiceBuilder,
1508    steps: Vec<BuilderStep>,
1509}
1510
1511impl OtherwiseBuilder {
1512    /// Close the otherwise scope and return the parent `ChoiceBuilder`.
1513    pub fn end_otherwise(self) -> ChoiceBuilder {
1514        let OtherwiseBuilder { mut parent, steps } = self;
1515        parent._otherwise = Some(steps);
1516        parent
1517    }
1518}
1519
1520impl StepAccumulator for OtherwiseBuilder {
1521    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1522        &mut self.steps
1523    }
1524}
1525
1526/// Builder for the sub-pipeline within a `.multicast()` ... `.end_multicast()` block.
1527///
1528/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1529/// but NOT `.build()` and NOT `.multicast()` (no nested multicasts).
1530///
1531/// Calling `.end_multicast()` packages the sub-steps into a `BuilderStep::Multicast`
1532/// and returns the parent `RouteBuilder`.
1533pub struct MulticastBuilder {
1534    parent: RouteBuilder,
1535    steps: Vec<BuilderStep>,
1536    config: MulticastConfig,
1537}
1538
1539impl MulticastBuilder {
1540    pub fn parallel(mut self, parallel: bool) -> Self {
1541        self.config = self.config.parallel(parallel);
1542        self
1543    }
1544
1545    pub fn parallel_limit(mut self, limit: usize) -> Self {
1546        self.config = self.config.parallel_limit(limit);
1547        self
1548    }
1549
1550    pub fn stop_on_exception(mut self, stop: bool) -> Self {
1551        self.config = self.config.stop_on_exception(stop);
1552        self
1553    }
1554
1555    pub fn timeout(mut self, duration: std::time::Duration) -> Self {
1556        self.config = self.config.timeout(duration);
1557        self
1558    }
1559
1560    pub fn aggregation(mut self, strategy: MulticastStrategy) -> Self {
1561        self.config = self.config.aggregation(strategy);
1562        self
1563    }
1564
1565    pub fn end_multicast(mut self) -> RouteBuilder {
1566        let step = BuilderStep::Multicast {
1567            steps: self.steps,
1568            config: self.config,
1569        };
1570        self.parent.steps.push(step);
1571        self.parent
1572    }
1573}
1574
1575impl StepAccumulator for MulticastBuilder {
1576    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1577        &mut self.steps
1578    }
1579}
1580
1581/// Builder for the sub-pipeline within a `.throttle()` ... `.end_throttle()` block.
1582///
1583/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1584/// but NOT `.build()` and NOT `.throttle()` (no nested throttles).
1585///
1586/// Calling `.end_throttle()` packages the sub-steps into a `BuilderStep::Throttle`
1587/// and returns the parent `RouteBuilder`.
1588pub struct ThrottleBuilder {
1589    parent: RouteBuilder,
1590    config: ThrottlerConfig,
1591    steps: Vec<BuilderStep>,
1592}
1593
1594impl ThrottleBuilder {
1595    /// Set the throttle strategy. Default is `Delay`.
1596    ///
1597    /// - `Delay`: Queue messages until capacity available
1598    /// - `Reject`: Return error immediately when throttled
1599    /// - `Drop`: Silently discard excess messages
1600    pub fn strategy(mut self, strategy: ThrottleStrategy) -> Self {
1601        self.config = self.config.strategy(strategy);
1602        self
1603    }
1604
1605    /// Close the throttle scope. Packages the accumulated sub-steps into a
1606    /// `BuilderStep::Throttle` and returns the parent `RouteBuilder`.
1607    pub fn end_throttle(mut self) -> RouteBuilder {
1608        let step = BuilderStep::Throttle {
1609            config: self.config,
1610            steps: self.steps,
1611        };
1612        self.parent.steps.push(step);
1613        self.parent
1614    }
1615}
1616
1617impl StepAccumulator for ThrottleBuilder {
1618    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1619        &mut self.steps
1620    }
1621}
1622
1623/// Builder for the sub-pipeline within a `.loop_count()` / `.loop_while()` ... `.end_loop()` block.
1624pub struct LoopBuilder {
1625    parent: RouteBuilder,
1626    config: LoopConfig,
1627    steps: Vec<BuilderStep>,
1628}
1629
1630impl LoopBuilder {
1631    pub fn loop_count(self, count: usize) -> LoopInLoopBuilder {
1632        LoopInLoopBuilder {
1633            parent: self,
1634            config: LoopConfig::new(LoopMode::Count(count)),
1635            steps: vec![],
1636        }
1637    }
1638
1639    pub fn loop_while<F>(self, predicate: F) -> LoopInLoopBuilder
1640    where
1641        F: Fn(&Exchange) -> bool + Send + Sync + 'static,
1642    {
1643        LoopInLoopBuilder {
1644            parent: self,
1645            config: LoopConfig::new(LoopMode::While(camel_api::FilterPredicate::new(predicate))),
1646            steps: vec![],
1647        }
1648    }
1649
1650    pub fn end_loop(mut self) -> RouteBuilder {
1651        let step = BuilderStep::Loop {
1652            config: self.config,
1653            steps: self.steps,
1654        };
1655        self.parent.steps.push(step);
1656        self.parent
1657    }
1658}
1659
1660impl StepAccumulator for LoopBuilder {
1661    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1662        &mut self.steps
1663    }
1664}
1665
1666pub struct LoopInLoopBuilder {
1667    parent: LoopBuilder,
1668    config: LoopConfig,
1669    steps: Vec<BuilderStep>,
1670}
1671
1672impl LoopInLoopBuilder {
1673    pub fn end_loop(mut self) -> LoopBuilder {
1674        let step = BuilderStep::Loop {
1675            config: self.config,
1676            steps: self.steps,
1677        };
1678        self.parent.steps.push(step);
1679        self.parent
1680    }
1681}
1682
1683impl StepAccumulator for LoopInLoopBuilder {
1684    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1685        &mut self.steps
1686    }
1687}
1688
1689/// Builder for the sub-pipeline within a `.load_balance()` ... `.end_load_balance()` block.
1690///
1691/// Exposes the same step methods as `RouteBuilder` (to, process, filter, etc.)
1692/// but NOT `.build()` and NOT `.load_balance()` (no nested load balancers).
1693///
1694/// Calling `.end_load_balance()` packages the sub-steps into a `BuilderStep::LoadBalance`
1695/// and returns the parent `RouteBuilder`.
1696pub struct LoadBalancerBuilder {
1697    parent: RouteBuilder,
1698    config: LoadBalancerConfig,
1699    steps: Vec<BuilderStep>,
1700}
1701
1702impl LoadBalancerBuilder {
1703    /// Set the load balance strategy to round-robin (default).
1704    pub fn round_robin(mut self) -> Self {
1705        self.config = LoadBalancerConfig::round_robin();
1706        self
1707    }
1708
1709    /// Set the load balance strategy to random selection.
1710    pub fn random(mut self) -> Self {
1711        self.config = LoadBalancerConfig::random();
1712        self
1713    }
1714
1715    /// Set the load balance strategy to weighted selection.
1716    ///
1717    /// Each endpoint is assigned a weight that determines its probability
1718    /// of being selected.
1719    pub fn weighted(mut self, weights: Vec<(String, u32)>) -> Self {
1720        self.config = LoadBalancerConfig::weighted(weights);
1721        self
1722    }
1723
1724    /// Set the load balance strategy to failover.
1725    ///
1726    /// Exchanges are sent to the first endpoint; on failure, the next endpoint
1727    /// is tried.
1728    pub fn failover(mut self) -> Self {
1729        self.config = LoadBalancerConfig::failover();
1730        self
1731    }
1732
1733    /// Close the load balance scope. Packages the accumulated sub-steps into a
1734    /// `BuilderStep::LoadBalance` and returns the parent `RouteBuilder`.
1735    pub fn end_load_balance(mut self) -> RouteBuilder {
1736        let step = BuilderStep::LoadBalance {
1737            config: self.config,
1738            steps: self.steps,
1739        };
1740        self.parent.steps.push(step);
1741        self.parent
1742    }
1743}
1744
1745impl StepAccumulator for LoadBalancerBuilder {
1746    fn steps_mut(&mut self) -> &mut Vec<BuilderStep> {
1747        &mut self.steps
1748    }
1749}
1750
1751// ---------------------------------------------------------------------------
1752// Tests
1753// ---------------------------------------------------------------------------
1754
1755#[cfg(test)]
1756mod tests {
1757    use super::*;
1758    use camel_api::SpanKindHint;
1759    use camel_api::error_handler::ErrorHandlerConfig;
1760    use camel_api::load_balancer::LoadBalanceStrategy;
1761    use camel_api::{Exchange, Message};
1762    use camel_core::route::BuilderStep;
1763    use std::sync::Arc;
1764    use std::time::Duration;
1765    use tower::{Service, ServiceExt};
1766
1767    #[test]
1768    fn test_builder_from_creates_definition() {
1769        let definition = RouteBuilder::from("timer:tick")
1770            .route_id("test-route")
1771            .build()
1772            .unwrap();
1773        assert_eq!(definition.from_uri(), "timer:tick");
1774    }
1775
1776    #[test]
1777    fn test_builder_empty_from_uri_errors() {
1778        let result = RouteBuilder::from("").route_id("test-route").build();
1779        assert!(result.is_err());
1780    }
1781
1782    #[test]
1783    fn test_build_rejects_schemeless_uri() {
1784        let result = RouteBuilder::from("no-scheme-here")
1785            .route_id("test-route")
1786            .build();
1787        match result {
1788            Err(err) => {
1789                let err_msg = format!("{err}");
1790                assert!(
1791                    err_msg.contains("scheme"),
1792                    "expected scheme-related error, got: {err_msg}"
1793                );
1794            }
1795            Ok(_) => panic!("schemeless URI should fail"),
1796        }
1797    }
1798
1799    #[test]
1800    fn test_build_rejects_empty_scheme_uri() {
1801        let result = RouteBuilder::from(":missing-scheme")
1802            .route_id("test-route")
1803            .build();
1804        match result {
1805            Err(err) => {
1806                let err_msg = format!("{err}");
1807                assert!(
1808                    err_msg.contains("scheme"),
1809                    "expected scheme-related error, got: {err_msg}"
1810                );
1811            }
1812            Ok(_) => panic!("empty-scheme URI should fail"),
1813        }
1814    }
1815
1816    #[test]
1817    fn test_build_accepts_valid_uri() {
1818        let result = RouteBuilder::from("timer:tick")
1819            .route_id("test-route")
1820            .build();
1821        assert!(result.is_ok());
1822    }
1823
1824    #[test]
1825    fn test_build_canonical_rejects_schemeless_uri() {
1826        let result = RouteBuilder::from("no-scheme-here")
1827            .route_id("test-route")
1828            .build_canonical();
1829        assert!(result.is_err());
1830    }
1831
1832    #[test]
1833    fn test_builder_to_adds_step() {
1834        let definition = RouteBuilder::from("timer:tick")
1835            .route_id("test-route")
1836            .to("log:info")
1837            .build()
1838            .unwrap();
1839
1840        assert_eq!(definition.from_uri(), "timer:tick");
1841        // We can verify steps were added by checking the structure
1842        assert!(matches!(&definition.steps()[0], BuilderStep::To(uri) if uri == "log:info"));
1843    }
1844
1845    #[test]
1846    fn test_builder_filter_adds_filter_step() {
1847        let definition = RouteBuilder::from("timer:tick")
1848            .route_id("test-route")
1849            .filter(|_ex| true)
1850            .to("mock:result")
1851            .end_filter()
1852            .build()
1853            .unwrap();
1854
1855        assert!(matches!(&definition.steps()[0], BuilderStep::Filter { .. }));
1856    }
1857
1858    #[test]
1859    fn test_builder_set_header_adds_processor_step() {
1860        let definition = RouteBuilder::from("timer:tick")
1861            .route_id("test-route")
1862            .set_header("key", Value::String("value".into()))
1863            .build()
1864            .unwrap();
1865
1866        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
1867    }
1868
1869    #[test]
1870    fn test_builder_map_body_adds_processor_step() {
1871        let definition = RouteBuilder::from("timer:tick")
1872            .route_id("test-route")
1873            .map_body(|body| body)
1874            .build()
1875            .unwrap();
1876
1877        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
1878    }
1879
1880    #[test]
1881    fn test_builder_process_adds_processor_step() {
1882        let definition = RouteBuilder::from("timer:tick")
1883            .route_id("test-route")
1884            .process(|ex| async move { Ok(ex) })
1885            .build()
1886            .unwrap();
1887
1888        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
1889    }
1890
1891    #[test]
1892    fn test_builder_chain_multiple_steps() {
1893        let definition = RouteBuilder::from("timer:tick")
1894            .route_id("test-route")
1895            .set_header("source", Value::String("timer".into()))
1896            .filter(|ex| ex.input.header("source").is_some())
1897            .to("log:info")
1898            .end_filter()
1899            .to("mock:result")
1900            .build()
1901            .unwrap();
1902
1903        assert_eq!(definition.steps().len(), 3); // set_header + Filter + To("mock:result")
1904        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_))); // set_header
1905        assert!(matches!(&definition.steps()[1], BuilderStep::Filter { .. })); // filter
1906        assert!(matches!(&definition.steps()[2], BuilderStep::To(uri) if uri == "mock:result"));
1907    }
1908
1909    #[test]
1910    fn test_loop_count_builder() {
1911        use camel_api::loop_eip::LoopMode;
1912
1913        let def = RouteBuilder::from("direct:start")
1914            .route_id("loop-test")
1915            .loop_count(3)
1916            .to("mock:inside")
1917            .end_loop()
1918            .to("mock:after")
1919            .build()
1920            .unwrap();
1921
1922        assert_eq!(def.steps().len(), 2);
1923        match &def.steps()[0] {
1924            BuilderStep::Loop { config, steps } => {
1925                assert!(matches!(config.mode, LoopMode::Count(3)));
1926                assert_eq!(steps.len(), 1);
1927            }
1928            other => panic!("Expected Loop, got {:?}", other),
1929        }
1930        assert!(matches!(def.steps()[1], BuilderStep::To(_)));
1931    }
1932
1933    #[test]
1934    fn test_loop_while_builder() {
1935        use camel_api::loop_eip::LoopMode;
1936
1937        let def = RouteBuilder::from("direct:start")
1938            .route_id("loop-while-test")
1939            .loop_while(|_ex| true)
1940            .to("mock:retry")
1941            .end_loop()
1942            .build()
1943            .unwrap();
1944
1945        assert_eq!(def.steps().len(), 1);
1946        match &def.steps()[0] {
1947            BuilderStep::Loop { config, steps } => {
1948                assert!(matches!(config.mode, LoopMode::While(_)));
1949                assert_eq!(steps.len(), 1);
1950            }
1951            other => panic!("Expected Loop, got {:?}", other),
1952        }
1953    }
1954
1955    #[test]
1956    fn test_nested_loop_builder() {
1957        use camel_api::loop_eip::LoopMode;
1958
1959        let def = RouteBuilder::from("direct:start")
1960            .route_id("nested-loop-test")
1961            .loop_count(2)
1962            .to("mock:outer")
1963            .loop_count(3)
1964            .to("mock:inner")
1965            .end_loop()
1966            .end_loop()
1967            .to("mock:after")
1968            .build()
1969            .unwrap();
1970
1971        assert_eq!(def.steps().len(), 2);
1972        match &def.steps()[0] {
1973            BuilderStep::Loop { steps, .. } => {
1974                assert_eq!(steps.len(), 2);
1975                match &steps[1] {
1976                    BuilderStep::Loop {
1977                        config,
1978                        steps: inner_steps,
1979                    } => {
1980                        assert!(matches!(config.mode, LoopMode::Count(3)));
1981                        assert_eq!(inner_steps.len(), 1);
1982                    }
1983                    other => panic!("Expected nested Loop, got {:?}", other),
1984                }
1985            }
1986            other => panic!("Expected outer Loop, got {:?}", other),
1987        }
1988    }
1989
1990    // -----------------------------------------------------------------------
1991    // Processor behavior tests — exercise the real Tower services directly
1992    // -----------------------------------------------------------------------
1993
1994    #[tokio::test]
1995    async fn test_set_header_processor_works() {
1996        let mut svc = SetHeader::new(IdentityProcessor, "greeting", Value::String("hello".into()));
1997        let exchange = Exchange::new(Message::new("test"));
1998        let result = svc.call(exchange).await.unwrap();
1999        assert_eq!(
2000            result.input.header("greeting"),
2001            Some(&Value::String("hello".into()))
2002        );
2003    }
2004
2005    #[tokio::test]
2006    async fn test_filter_processor_passes() {
2007        use camel_api::BoxProcessorExt;
2008        use camel_processor::FilterService;
2009
2010        let sub = BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }));
2011        let mut svc =
2012            FilterService::new(|ex: &Exchange| ex.input.body.as_text() == Some("pass"), sub);
2013        let exchange = Exchange::new(Message::new("pass"));
2014        let result = svc.ready().await.unwrap().call(exchange).await.unwrap();
2015        assert_eq!(result.input.body.as_text(), Some("pass"));
2016    }
2017
2018    #[tokio::test]
2019    async fn test_filter_processor_blocks() {
2020        use camel_api::BoxProcessorExt;
2021        use camel_processor::FilterService;
2022
2023        let sub = BoxProcessor::from_fn(|_ex| {
2024            Box::pin(async move { Err(CamelError::ProcessorError("should not reach".into())) })
2025        });
2026        let mut svc =
2027            FilterService::new(|ex: &Exchange| ex.input.body.as_text() == Some("pass"), sub);
2028        let exchange = Exchange::new(Message::new("reject"));
2029        let result = svc.ready().await.unwrap().call(exchange).await.unwrap();
2030        assert_eq!(result.input.body.as_text(), Some("reject"));
2031    }
2032
2033    #[tokio::test]
2034    async fn test_map_body_processor_works() {
2035        let mapper = MapBody::new(IdentityProcessor, |body: Body| {
2036            if let Some(text) = body.as_text() {
2037                Body::Text(text.to_uppercase())
2038            } else {
2039                body
2040            }
2041        });
2042        let exchange = Exchange::new(Message::new("hello"));
2043        let result = mapper.oneshot(exchange).await.unwrap();
2044        assert_eq!(result.input.body.as_text(), Some("HELLO"));
2045    }
2046
2047    #[tokio::test]
2048    async fn test_process_custom_processor_works() {
2049        let processor = ProcessorFn::new(|mut ex: Exchange| async move {
2050            ex.set_property("custom", Value::Bool(true));
2051            Ok(ex)
2052        });
2053        let exchange = Exchange::new(Message::default());
2054        let result = processor.oneshot(exchange).await.unwrap();
2055        assert_eq!(result.property("custom"), Some(&Value::Bool(true)));
2056    }
2057
2058    // -----------------------------------------------------------------------
2059    // Sequential pipeline test
2060    // -----------------------------------------------------------------------
2061
2062    #[tokio::test]
2063    async fn test_compose_pipeline_runs_steps_in_order() {
2064        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2065
2066        let processors = vec![
2067            CompiledStep::Process {
2068                kind_hint: SpanKindHint::Internal,
2069                processor: BoxProcessor::new(SetHeader::new(
2070                    IdentityProcessor,
2071                    "step",
2072                    Value::String("one".into()),
2073                )),
2074                body_contract: None,
2075                lifecycle: None,
2076                label: None,
2077            },
2078            CompiledStep::Process {
2079                kind_hint: SpanKindHint::Internal,
2080                processor: BoxProcessor::new(MapBody::new(IdentityProcessor, |body: Body| {
2081                    if let Some(text) = body.as_text() {
2082                        Body::Text(format!("{}-processed", text))
2083                    } else {
2084                        body
2085                    }
2086                })),
2087                body_contract: None,
2088                lifecycle: None,
2089                label: None,
2090            },
2091        ];
2092
2093        let pipeline = compose_pipeline(processors, PipelineRuntimeCtx::compile_time());
2094        let exchange = Exchange::new(Message::new("hello"));
2095        let result = pipeline.oneshot(exchange).await.unwrap();
2096
2097        assert_eq!(
2098            result.input.header("step"),
2099            Some(&Value::String("one".into()))
2100        );
2101        assert_eq!(result.input.body.as_text(), Some("hello-processed"));
2102    }
2103
2104    #[tokio::test]
2105    async fn test_compose_pipeline_empty_is_identity() {
2106        use camel_core::route::{PipelineRuntimeCtx, compose_pipeline};
2107
2108        let pipeline = compose_pipeline(vec![], PipelineRuntimeCtx::compile_time());
2109        let exchange = Exchange::new(Message::new("unchanged"));
2110        let result = pipeline.oneshot(exchange).await.unwrap();
2111        assert_eq!(result.input.body.as_text(), Some("unchanged"));
2112    }
2113
2114    // -----------------------------------------------------------------------
2115    // Circuit breaker builder tests
2116    // -----------------------------------------------------------------------
2117
2118    #[test]
2119    fn test_builder_circuit_breaker_sets_config() {
2120        use camel_api::circuit_breaker::CircuitBreakerConfig;
2121
2122        let config = CircuitBreakerConfig::new().failure_threshold(5);
2123        let definition = RouteBuilder::from("timer:tick")
2124            .route_id("test-route")
2125            .circuit_breaker(config)
2126            .build()
2127            .unwrap();
2128
2129        let cb = definition
2130            .circuit_breaker_config()
2131            .expect("circuit breaker should be set");
2132        assert_eq!(cb.failure_threshold, 5);
2133    }
2134
2135    #[test]
2136    fn test_builder_circuit_breaker_with_error_handler() {
2137        use camel_api::circuit_breaker::CircuitBreakerConfig;
2138        use camel_api::error_handler::ErrorHandlerConfig;
2139
2140        let cb_config = CircuitBreakerConfig::new().failure_threshold(3);
2141        let eh_config = ErrorHandlerConfig::log_only();
2142
2143        let definition = RouteBuilder::from("timer:tick")
2144            .route_id("test-route")
2145            .to("log:info")
2146            .circuit_breaker(cb_config)
2147            .error_handler(eh_config)
2148            .build()
2149            .unwrap();
2150
2151        assert!(
2152            definition.circuit_breaker_config().is_some(),
2153            "circuit breaker config should be set"
2154        );
2155        // Route definition was built successfully with both configs.
2156    }
2157
2158    #[test]
2159    fn test_builder_on_exception_shorthand_multiple_clauses_preserve_order() {
2160        let definition = RouteBuilder::from("direct:start")
2161            .route_id("test-route")
2162            .dead_letter_channel("log:dlc")
2163            .on_exception(|e| matches!(e, CamelError::Io(_)))
2164            .retry(3)
2165            .handled_by("log:io")
2166            .end_on_exception()
2167            .on_exception(|e| matches!(e, CamelError::ProcessorError(_)))
2168            .retry(1)
2169            .end_on_exception()
2170            .to("mock:out")
2171            .build()
2172            .expect("route should build");
2173
2174        let cfg = definition
2175            .error_handler_config()
2176            .expect("error handler should be set");
2177        assert_eq!(cfg.policies.len(), 2);
2178        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:dlc"));
2179        assert_eq!(
2180            cfg.policies[0].retry.as_ref().map(|p| p.max_attempts),
2181            Some(3)
2182        );
2183        assert_eq!(cfg.policies[0].handled_by.as_deref(), Some("log:io"));
2184        assert_eq!(
2185            cfg.policies[1].retry.as_ref().map(|p| p.max_attempts),
2186            Some(1)
2187        );
2188    }
2189
2190    #[test]
2191    fn test_builder_on_exception_mixed_mode_rejected() {
2192        let result = RouteBuilder::from("direct:start")
2193            .route_id("test-route")
2194            .error_handler(ErrorHandlerConfig::log_only())
2195            .on_exception(|_e| true)
2196            .end_on_exception()
2197            .to("mock:out")
2198            .build();
2199
2200        let err = result.err().expect("mixed mode should fail with an error");
2201
2202        assert!(
2203            format!("{err}").contains("mixed error handler modes"),
2204            "unexpected error: {err}"
2205        );
2206    }
2207
2208    #[test]
2209    fn test_builder_on_exception_backoff_and_jitter_without_retry_noop() {
2210        let definition = RouteBuilder::from("direct:start")
2211            .route_id("test-route")
2212            .on_exception(|_e| true)
2213            .with_backoff(Duration::from_millis(5), 3.0, Duration::from_millis(100))
2214            .with_jitter(0.5)
2215            .end_on_exception()
2216            .to("mock:out")
2217            .build()
2218            .expect("route should build");
2219
2220        let cfg = definition
2221            .error_handler_config()
2222            .expect("error handler should be set");
2223        assert_eq!(cfg.policies.len(), 1);
2224        assert!(cfg.policies[0].retry.is_none());
2225    }
2226
2227    #[test]
2228    fn test_builder_dead_letter_channel_without_on_exception_sets_dlc() {
2229        let definition = RouteBuilder::from("direct:start")
2230            .route_id("test-route")
2231            .dead_letter_channel("log:dlc")
2232            .to("mock:out")
2233            .build()
2234            .expect("route should build");
2235
2236        let cfg = definition
2237            .error_handler_config()
2238            .expect("error handler should be set");
2239        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:dlc"));
2240        assert!(cfg.policies.is_empty());
2241    }
2242
2243    #[test]
2244    fn test_builder_dead_letter_channel_called_twice_uses_latest_and_keeps_policies() {
2245        let definition = RouteBuilder::from("direct:start")
2246            .route_id("test-route")
2247            .dead_letter_channel("log:first")
2248            .on_exception(|e| matches!(e, CamelError::Io(_)))
2249            .retry(2)
2250            .end_on_exception()
2251            .dead_letter_channel("log:second")
2252            .to("mock:out")
2253            .build()
2254            .expect("route should build");
2255
2256        let cfg = definition
2257            .error_handler_config()
2258            .expect("error handler should be set");
2259        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:second"));
2260        assert_eq!(cfg.policies.len(), 1);
2261        assert_eq!(
2262            cfg.policies[0].retry.as_ref().map(|p| p.max_attempts),
2263            Some(2)
2264        );
2265    }
2266
2267    #[test]
2268    fn test_builder_on_exception_without_dlc_defaults_to_log_only() {
2269        let definition = RouteBuilder::from("direct:start")
2270            .route_id("test-route")
2271            .on_exception(|e| matches!(e, CamelError::ProcessorError(_)))
2272            .retry(1)
2273            .end_on_exception()
2274            .to("mock:out")
2275            .build()
2276            .expect("route should build");
2277
2278        let cfg = definition
2279            .error_handler_config()
2280            .expect("error handler should be set");
2281        assert!(cfg.dlc_uri.is_none());
2282        assert_eq!(cfg.policies.len(), 1);
2283    }
2284
2285    #[test]
2286    fn test_builder_error_handler_explicit_overwrite_stays_explicit_mode() {
2287        let first = ErrorHandlerConfig::dead_letter_channel("log:first");
2288        let second = ErrorHandlerConfig::dead_letter_channel("log:second");
2289
2290        let definition = RouteBuilder::from("direct:start")
2291            .route_id("test-route")
2292            .error_handler(first)
2293            .error_handler(second)
2294            .to("mock:out")
2295            .build()
2296            .expect("route should build");
2297
2298        let cfg = definition
2299            .error_handler_config()
2300            .expect("error handler should be set");
2301        assert_eq!(cfg.dlc_uri.as_deref(), Some("log:second"));
2302    }
2303
2304    // --- Splitter builder tests ---
2305
2306    #[test]
2307    fn test_split_builder_typestate() {
2308        use camel_api::splitter::{SplitterConfig, split_body_lines};
2309
2310        // .split() returns SplitBuilder, .end_split() returns RouteBuilder
2311        let definition = RouteBuilder::from("timer:test?period=1000")
2312            .route_id("test-route")
2313            .split(SplitterConfig::new(split_body_lines()))
2314            .to("mock:per-fragment")
2315            .end_split()
2316            .to("mock:final")
2317            .build()
2318            .unwrap();
2319
2320        // Should have 2 top-level steps: Split + To("mock:final")
2321        assert_eq!(definition.steps().len(), 2);
2322    }
2323
2324    #[test]
2325    fn test_split_builder_steps_collected() {
2326        use camel_api::splitter::{SplitterConfig, split_body_lines};
2327
2328        let definition = RouteBuilder::from("timer:test?period=1000")
2329            .route_id("test-route")
2330            .split(SplitterConfig::new(split_body_lines()))
2331            .set_header("fragment", Value::String("yes".into()))
2332            .to("mock:per-fragment")
2333            .end_split()
2334            .build()
2335            .unwrap();
2336
2337        // Should have 1 top-level step: Split (containing 2 sub-steps)
2338        assert_eq!(definition.steps().len(), 1);
2339        match &definition.steps()[0] {
2340            BuilderStep::Split { steps, .. } => {
2341                assert_eq!(steps.len(), 2); // SetHeader + To
2342            }
2343            other => panic!("Expected Split, got {:?}", other),
2344        }
2345    }
2346
2347    #[test]
2348    fn test_split_builder_config_propagated() {
2349        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
2350
2351        let definition = RouteBuilder::from("timer:test?period=1000")
2352            .route_id("test-route")
2353            .split(
2354                SplitterConfig::new(split_body_lines())
2355                    .parallel(true)
2356                    .parallel_limit(4)
2357                    .aggregation(AggregationStrategy::CollectAll),
2358            )
2359            .to("mock:per-fragment")
2360            .end_split()
2361            .build()
2362            .unwrap();
2363
2364        match &definition.steps()[0] {
2365            BuilderStep::Split { config, .. } => {
2366                assert!(config.parallel);
2367                assert_eq!(config.parallel_limit, Some(4));
2368                assert!(matches!(
2369                    config.aggregation,
2370                    AggregationStrategy::CollectAll
2371                ));
2372            }
2373            other => panic!("Expected Split, got {:?}", other),
2374        }
2375    }
2376
2377    #[test]
2378    fn test_aggregate_builder_adds_step() {
2379        use camel_api::aggregator::AggregatorConfig;
2380        use camel_core::route::BuilderStep;
2381
2382        let definition = RouteBuilder::from("timer:tick")
2383            .route_id("test-route")
2384            .aggregate(
2385                AggregatorConfig::correlate_by("key")
2386                    .complete_when_size(2)
2387                    .build()
2388                    .unwrap(),
2389            )
2390            .build()
2391            .unwrap();
2392
2393        assert_eq!(definition.steps().len(), 1);
2394        assert!(matches!(
2395            definition.steps()[0],
2396            BuilderStep::Aggregate { .. }
2397        ));
2398    }
2399
2400    #[test]
2401    fn test_aggregate_in_split_builder() {
2402        use camel_api::aggregator::AggregatorConfig;
2403        use camel_api::splitter::{SplitterConfig, split_body_lines};
2404        use camel_core::route::BuilderStep;
2405
2406        let definition = RouteBuilder::from("timer:tick")
2407            .route_id("test-route")
2408            .split(SplitterConfig::new(split_body_lines()))
2409            .aggregate(
2410                AggregatorConfig::correlate_by("key")
2411                    .complete_when_size(1)
2412                    .build()
2413                    .unwrap(),
2414            )
2415            .end_split()
2416            .build()
2417            .unwrap();
2418
2419        assert_eq!(definition.steps().len(), 1);
2420        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
2421            assert!(matches!(steps[0], BuilderStep::Aggregate { .. }));
2422        } else {
2423            panic!("expected Split step");
2424        }
2425    }
2426
2427    // ── set_body / set_body_fn / set_header_fn builder tests ────────────────────
2428
2429    #[test]
2430    fn test_builder_set_body_static_adds_processor() {
2431        let definition = RouteBuilder::from("timer:tick")
2432            .route_id("test-route")
2433            .set_body("fixed")
2434            .build()
2435            .unwrap();
2436        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
2437    }
2438
2439    #[test]
2440    fn test_builder_set_body_fn_adds_processor() {
2441        let definition = RouteBuilder::from("timer:tick")
2442            .route_id("test-route")
2443            .set_body_fn(|_ex: &Exchange| Body::Text("dynamic".into()))
2444            .build()
2445            .unwrap();
2446        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
2447    }
2448
2449    #[test]
2450    fn transform_alias_produces_same_as_set_body() {
2451        let route_transform = RouteBuilder::from("timer:tick")
2452            .route_id("test-route")
2453            .transform("hello")
2454            .build()
2455            .unwrap();
2456
2457        let route_set_body = RouteBuilder::from("timer:tick")
2458            .route_id("test-route")
2459            .set_body("hello")
2460            .build()
2461            .unwrap();
2462
2463        assert_eq!(route_transform.steps().len(), route_set_body.steps().len());
2464    }
2465
2466    #[test]
2467    fn test_builder_set_header_fn_adds_processor() {
2468        let definition = RouteBuilder::from("timer:tick")
2469            .route_id("test-route")
2470            .set_header_fn("k", |_ex: &Exchange| Value::String("v".into()))
2471            .build()
2472            .unwrap();
2473        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
2474    }
2475
2476    #[tokio::test]
2477    async fn test_set_body_static_processor_works() {
2478        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2479        let def = RouteBuilder::from("t:t")
2480            .route_id("test-route")
2481            .set_body("replaced")
2482            .build()
2483            .unwrap();
2484        let pipeline = compose_pipeline(
2485            def.steps()
2486                .iter()
2487                .filter_map(|s| {
2488                    if let BuilderStep::Processor(op) = s {
2489                        Some(op.0.clone())
2490                    } else {
2491                        None
2492                    }
2493                })
2494                .map(|p| CompiledStep::Process {
2495                    kind_hint: SpanKindHint::Internal,
2496                    processor: p,
2497                    body_contract: None,
2498                    lifecycle: None,
2499                    label: None,
2500                })
2501                .collect(),
2502            PipelineRuntimeCtx::compile_time(),
2503        );
2504        let exchange = Exchange::new(Message::new("original"));
2505        let result = pipeline.oneshot(exchange).await.unwrap();
2506        assert_eq!(result.input.body.as_text(), Some("replaced"));
2507    }
2508
2509    #[tokio::test]
2510    async fn test_set_body_fn_processor_works() {
2511        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2512        let def = RouteBuilder::from("t:t")
2513            .route_id("test-route")
2514            .set_body_fn(|ex: &Exchange| {
2515                Body::Text(ex.input.body.as_text().unwrap_or("").to_uppercase())
2516            })
2517            .build()
2518            .unwrap();
2519        let pipeline = compose_pipeline(
2520            def.steps()
2521                .iter()
2522                .filter_map(|s| {
2523                    if let BuilderStep::Processor(op) = s {
2524                        Some(op.0.clone())
2525                    } else {
2526                        None
2527                    }
2528                })
2529                .map(|p| CompiledStep::Process {
2530                    kind_hint: SpanKindHint::Internal,
2531                    processor: p,
2532                    body_contract: None,
2533                    lifecycle: None,
2534                    label: None,
2535                })
2536                .collect(),
2537            PipelineRuntimeCtx::compile_time(),
2538        );
2539        let exchange = Exchange::new(Message::new("hello"));
2540        let result = pipeline.oneshot(exchange).await.unwrap();
2541        assert_eq!(result.input.body.as_text(), Some("HELLO"));
2542    }
2543
2544    #[tokio::test]
2545    async fn test_set_header_fn_processor_works() {
2546        use camel_core::route::{CompiledStep, PipelineRuntimeCtx, compose_pipeline};
2547        let def = RouteBuilder::from("t:t")
2548            .route_id("test-route")
2549            .set_header_fn("echo", |ex: &Exchange| {
2550                ex.input
2551                    .body
2552                    .as_text()
2553                    .map(|t| Value::String(t.into()))
2554                    .unwrap_or(Value::Null)
2555            })
2556            .build()
2557            .unwrap();
2558        let pipeline = compose_pipeline(
2559            def.steps()
2560                .iter()
2561                .filter_map(|s| {
2562                    if let BuilderStep::Processor(op) = s {
2563                        Some(op.0.clone())
2564                    } else {
2565                        None
2566                    }
2567                })
2568                .map(|p| CompiledStep::Process {
2569                    kind_hint: SpanKindHint::Internal,
2570                    processor: p,
2571                    body_contract: None,
2572                    lifecycle: None,
2573                    label: None,
2574                })
2575                .collect(),
2576            PipelineRuntimeCtx::compile_time(),
2577        );
2578        let exchange = Exchange::new(Message::new("ping"));
2579        let result = pipeline.oneshot(exchange).await.unwrap();
2580        assert_eq!(
2581            result.input.header("echo"),
2582            Some(&Value::String("ping".into()))
2583        );
2584    }
2585
2586    // ── FilterBuilder typestate tests ─────────────────────────────────────
2587
2588    #[test]
2589    fn test_filter_builder_typestate() {
2590        let result = RouteBuilder::from("timer:tick?period=50&repeatCount=1")
2591            .route_id("test-route")
2592            .filter(|_ex| true)
2593            .to("mock:inner")
2594            .end_filter()
2595            .to("mock:outer")
2596            .build();
2597        assert!(result.is_ok());
2598    }
2599
2600    #[test]
2601    fn test_filter_builder_steps_collected() {
2602        let definition = RouteBuilder::from("timer:tick?period=50&repeatCount=1")
2603            .route_id("test-route")
2604            .filter(|_ex| true)
2605            .to("mock:inner")
2606            .end_filter()
2607            .build()
2608            .unwrap();
2609
2610        assert_eq!(definition.steps().len(), 1);
2611        assert!(matches!(&definition.steps()[0], BuilderStep::Filter { .. }));
2612    }
2613
2614    #[test]
2615    fn test_wire_tap_builder_adds_step() {
2616        let definition = RouteBuilder::from("timer:tick")
2617            .route_id("test-route")
2618            .wire_tap("mock:tap")
2619            .to("mock:result")
2620            .build()
2621            .unwrap();
2622
2623        assert_eq!(definition.steps().len(), 2);
2624        assert!(
2625            matches!(&definition.steps()[0], BuilderStep::WireTap { uri } if uri == "mock:tap")
2626        );
2627        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:result"));
2628    }
2629
2630    // ── MulticastBuilder typestate tests ─────────────────────────────────────
2631
2632    #[test]
2633    fn test_multicast_builder_typestate() {
2634        let definition = RouteBuilder::from("timer:tick")
2635            .route_id("test-route")
2636            .multicast()
2637            .to("direct:a")
2638            .to("direct:b")
2639            .end_multicast()
2640            .to("mock:result")
2641            .build()
2642            .unwrap();
2643
2644        assert_eq!(definition.steps().len(), 2); // Multicast + To("mock:result")
2645    }
2646
2647    #[test]
2648    fn test_multicast_builder_steps_collected() {
2649        let definition = RouteBuilder::from("timer:tick")
2650            .route_id("test-route")
2651            .multicast()
2652            .to("direct:a")
2653            .to("direct:b")
2654            .end_multicast()
2655            .build()
2656            .unwrap();
2657
2658        match &definition.steps()[0] {
2659            BuilderStep::Multicast { steps, .. } => {
2660                assert_eq!(steps.len(), 2);
2661            }
2662            other => panic!("Expected Multicast, got {:?}", other),
2663        }
2664    }
2665
2666    // ── Concurrency builder tests ─────────────────────────────────────
2667
2668    #[test]
2669    fn test_builder_concurrent_sets_concurrency() {
2670        use camel_component_api::ConcurrencyModel;
2671
2672        let definition = RouteBuilder::from("http://0.0.0.0:8080/test")
2673            .route_id("test-route")
2674            .concurrent(16)
2675            .to("log:info")
2676            .build()
2677            .unwrap();
2678
2679        assert_eq!(
2680            definition.concurrency_override(),
2681            Some(&ConcurrencyModel::Concurrent { max: Some(16) })
2682        );
2683    }
2684
2685    #[test]
2686    fn test_builder_concurrent_zero_means_unbounded() {
2687        use camel_component_api::ConcurrencyModel;
2688
2689        let definition = RouteBuilder::from("http://0.0.0.0:8080/test")
2690            .route_id("test-route")
2691            .concurrent(0)
2692            .to("log:info")
2693            .build()
2694            .unwrap();
2695
2696        assert_eq!(
2697            definition.concurrency_override(),
2698            Some(&ConcurrencyModel::Concurrent { max: None })
2699        );
2700    }
2701
2702    #[test]
2703    fn test_builder_sequential_sets_concurrency() {
2704        use camel_component_api::ConcurrencyModel;
2705
2706        let definition = RouteBuilder::from("http://0.0.0.0:8080/test")
2707            .route_id("test-route")
2708            .sequential()
2709            .to("log:info")
2710            .build()
2711            .unwrap();
2712
2713        assert_eq!(
2714            definition.concurrency_override(),
2715            Some(&ConcurrencyModel::Sequential)
2716        );
2717    }
2718
2719    #[test]
2720    fn test_builder_default_concurrency_is_none() {
2721        let definition = RouteBuilder::from("timer:tick")
2722            .route_id("test-route")
2723            .to("log:info")
2724            .build()
2725            .unwrap();
2726
2727        assert_eq!(definition.concurrency_override(), None);
2728    }
2729
2730    // ── Route lifecycle builder tests ─────────────────────────────────────
2731
2732    #[test]
2733    fn test_builder_route_id_sets_id() {
2734        let definition = RouteBuilder::from("timer:tick")
2735            .route_id("my-route")
2736            .build()
2737            .unwrap();
2738
2739        assert_eq!(definition.route_id(), "my-route");
2740    }
2741
2742    #[test]
2743    fn test_build_without_route_id_fails() {
2744        let result = RouteBuilder::from("timer:tick?period=1000")
2745            .to("log:info")
2746            .build();
2747        let err = match result {
2748            Err(e) => e.to_string(),
2749            Ok(_) => panic!("build() should fail without route_id"),
2750        };
2751        assert!(
2752            err.contains("route_id"),
2753            "error should mention route_id, got: {}",
2754            err
2755        );
2756    }
2757
2758    #[test]
2759    fn test_builder_empty_route_id_rejected() {
2760        let result = RouteBuilder::from("timer:tick").route_id("").build();
2761        let err = result.err().expect("empty route_id should be rejected");
2762        assert!(matches!(err, CamelError::RouteError(_)));
2763    }
2764
2765    #[test]
2766    fn test_builder_whitespace_route_id_rejected() {
2767        let result = RouteBuilder::from("timer:tick").route_id("   ").build();
2768        assert!(result.is_err());
2769    }
2770
2771    #[test]
2772    fn test_builder_auto_startup_false() {
2773        let definition = RouteBuilder::from("timer:tick")
2774            .route_id("test-route")
2775            .auto_startup(false)
2776            .build()
2777            .unwrap();
2778
2779        assert!(!definition.auto_startup());
2780    }
2781
2782    #[test]
2783    fn test_builder_startup_order_custom() {
2784        let definition = RouteBuilder::from("timer:tick")
2785            .route_id("test-route")
2786            .startup_order(50)
2787            .build()
2788            .unwrap();
2789
2790        assert_eq!(definition.startup_order(), 50);
2791    }
2792
2793    #[test]
2794    fn test_builder_defaults() {
2795        let definition = RouteBuilder::from("timer:tick")
2796            .route_id("test-route")
2797            .build()
2798            .unwrap();
2799
2800        assert_eq!(definition.route_id(), "test-route");
2801        assert!(definition.auto_startup());
2802        assert_eq!(definition.startup_order(), 1000);
2803    }
2804
2805    // ── Choice typestate tests ──────────────────────────────────────────────────
2806
2807    #[test]
2808    fn test_choice_builder_single_when() {
2809        let definition = RouteBuilder::from("timer:tick")
2810            .route_id("test-route")
2811            .choice()
2812            .when(|ex: &Exchange| ex.input.header("type").is_some())
2813            .to("mock:typed")
2814            .end_when()
2815            .end_choice()
2816            .build()
2817            .unwrap();
2818        assert_eq!(definition.steps().len(), 1);
2819        assert!(
2820            matches!(&definition.steps()[0], BuilderStep::Choice { whens, otherwise }
2821            if whens.len() == 1 && otherwise.is_none())
2822        );
2823    }
2824
2825    #[test]
2826    fn test_choice_builder_when_otherwise() {
2827        let definition = RouteBuilder::from("timer:tick")
2828            .route_id("test-route")
2829            .choice()
2830            .when(|ex: &Exchange| ex.input.header("a").is_some())
2831            .to("mock:a")
2832            .end_when()
2833            .otherwise()
2834            .to("mock:fallback")
2835            .end_otherwise()
2836            .end_choice()
2837            .build()
2838            .unwrap();
2839        assert!(
2840            matches!(&definition.steps()[0], BuilderStep::Choice { whens, otherwise }
2841            if whens.len() == 1 && otherwise.is_some())
2842        );
2843    }
2844
2845    #[test]
2846    fn test_choice_builder_multiple_whens() {
2847        let definition = RouteBuilder::from("timer:tick")
2848            .route_id("test-route")
2849            .choice()
2850            .when(|ex: &Exchange| ex.input.header("a").is_some())
2851            .to("mock:a")
2852            .end_when()
2853            .when(|ex: &Exchange| ex.input.header("b").is_some())
2854            .to("mock:b")
2855            .end_when()
2856            .end_choice()
2857            .build()
2858            .unwrap();
2859        assert!(
2860            matches!(&definition.steps()[0], BuilderStep::Choice { whens, .. }
2861            if whens.len() == 2)
2862        );
2863    }
2864
2865    #[test]
2866    fn test_choice_step_after_choice() {
2867        // Steps after end_choice() are added to the outer pipeline, not inside choice.
2868        let definition = RouteBuilder::from("timer:tick")
2869            .route_id("test-route")
2870            .choice()
2871            .when(|_ex: &Exchange| true)
2872            .to("mock:inner")
2873            .end_when()
2874            .end_choice()
2875            .to("mock:outer") // must be step[1], not inside choice
2876            .build()
2877            .unwrap();
2878        assert_eq!(definition.steps().len(), 2);
2879        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
2880    }
2881
2882    // ── Throttle typestate tests ──────────────────────────────────────────────────
2883
2884    #[test]
2885    fn test_throttle_builder_typestate() {
2886        let definition = RouteBuilder::from("timer:tick")
2887            .route_id("test-route")
2888            .throttle(10, std::time::Duration::from_secs(1))
2889            .to("mock:result")
2890            .end_throttle()
2891            .build()
2892            .unwrap();
2893
2894        assert_eq!(definition.steps().len(), 1);
2895        assert!(matches!(
2896            &definition.steps()[0],
2897            BuilderStep::Throttle { .. }
2898        ));
2899    }
2900
2901    #[test]
2902    fn test_throttle_builder_with_strategy() {
2903        let definition = RouteBuilder::from("timer:tick")
2904            .route_id("test-route")
2905            .throttle(10, std::time::Duration::from_secs(1))
2906            .strategy(ThrottleStrategy::Reject)
2907            .to("mock:result")
2908            .end_throttle()
2909            .build()
2910            .unwrap();
2911
2912        if let BuilderStep::Throttle { config, .. } = &definition.steps()[0] {
2913            assert_eq!(config.strategy, ThrottleStrategy::Reject);
2914        } else {
2915            panic!("Expected Throttle step");
2916        }
2917    }
2918
2919    #[test]
2920    fn test_throttle_builder_steps_collected() {
2921        let definition = RouteBuilder::from("timer:tick")
2922            .route_id("test-route")
2923            .throttle(5, std::time::Duration::from_secs(1))
2924            .set_header("throttled", Value::Bool(true))
2925            .to("mock:throttled")
2926            .end_throttle()
2927            .build()
2928            .unwrap();
2929
2930        match &definition.steps()[0] {
2931            BuilderStep::Throttle { steps, .. } => {
2932                assert_eq!(steps.len(), 2); // SetHeader + To
2933            }
2934            other => panic!("Expected Throttle, got {:?}", other),
2935        }
2936    }
2937
2938    #[test]
2939    fn test_throttle_step_after_throttle() {
2940        // Steps after end_throttle() are added to the outer pipeline, not inside throttle.
2941        let definition = RouteBuilder::from("timer:tick")
2942            .route_id("test-route")
2943            .throttle(10, std::time::Duration::from_secs(1))
2944            .to("mock:inner")
2945            .end_throttle()
2946            .to("mock:outer")
2947            .build()
2948            .unwrap();
2949
2950        assert_eq!(definition.steps().len(), 2);
2951        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
2952    }
2953
2954    // ── LoadBalance typestate tests ──────────────────────────────────────────────────
2955
2956    #[test]
2957    fn test_load_balance_builder_typestate() {
2958        let definition = RouteBuilder::from("timer:tick")
2959            .route_id("test-route")
2960            .load_balance()
2961            .round_robin()
2962            .to("mock:a")
2963            .to("mock:b")
2964            .end_load_balance()
2965            .build()
2966            .unwrap();
2967
2968        assert_eq!(definition.steps().len(), 1);
2969        assert!(matches!(
2970            &definition.steps()[0],
2971            BuilderStep::LoadBalance { .. }
2972        ));
2973    }
2974
2975    #[test]
2976    fn test_load_balance_builder_with_strategy() {
2977        let definition = RouteBuilder::from("timer:tick")
2978            .route_id("test-route")
2979            .load_balance()
2980            .random()
2981            .to("mock:result")
2982            .end_load_balance()
2983            .build()
2984            .unwrap();
2985
2986        if let BuilderStep::LoadBalance { config, .. } = &definition.steps()[0] {
2987            assert_eq!(config.strategy, LoadBalanceStrategy::Random);
2988        } else {
2989            panic!("Expected LoadBalance step");
2990        }
2991    }
2992
2993    #[test]
2994    fn test_load_balance_builder_steps_collected() {
2995        let definition = RouteBuilder::from("timer:tick")
2996            .route_id("test-route")
2997            .load_balance()
2998            .set_header("lb", Value::Bool(true))
2999            .to("mock:a")
3000            .end_load_balance()
3001            .build()
3002            .unwrap();
3003
3004        match &definition.steps()[0] {
3005            BuilderStep::LoadBalance { steps, .. } => {
3006                assert_eq!(steps.len(), 2); // SetHeader + To
3007            }
3008            other => panic!("Expected LoadBalance, got {:?}", other),
3009        }
3010    }
3011
3012    #[test]
3013    fn test_load_balance_step_after_load_balance() {
3014        // Steps after end_load_balance() are added to the outer pipeline, not inside load_balance.
3015        let definition = RouteBuilder::from("timer:tick")
3016            .route_id("test-route")
3017            .load_balance()
3018            .to("mock:inner")
3019            .end_load_balance()
3020            .to("mock:outer")
3021            .build()
3022            .unwrap();
3023
3024        assert_eq!(definition.steps().len(), 2);
3025        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
3026    }
3027
3028    // ── DynamicRouter typestate tests ──────────────────────────────────────────────────
3029
3030    #[test]
3031    fn test_dynamic_router_builder() {
3032        let definition = RouteBuilder::from("timer:tick")
3033            .route_id("test-route")
3034            .dynamic_router(Arc::new(|_| Some("mock:result".to_string())))
3035            .build()
3036            .unwrap();
3037
3038        assert_eq!(definition.steps().len(), 1);
3039        assert!(matches!(
3040            &definition.steps()[0],
3041            BuilderStep::DynamicRouter { .. }
3042        ));
3043    }
3044
3045    #[test]
3046    fn test_dynamic_router_builder_with_config() {
3047        let config = DynamicRouterConfig::new(Arc::new(|_| Some("mock:a".to_string())))
3048            .max_iterations(100)
3049            .cache_size(500);
3050
3051        let definition = RouteBuilder::from("timer:tick")
3052            .route_id("test-route")
3053            .dynamic_router_with_config(config)
3054            .build()
3055            .unwrap();
3056
3057        assert_eq!(definition.steps().len(), 1);
3058        if let BuilderStep::DynamicRouter { config } = &definition.steps()[0] {
3059            assert_eq!(config.max_iterations, 100);
3060            assert_eq!(config.cache_size, 500);
3061        } else {
3062            panic!("Expected DynamicRouter step");
3063        }
3064    }
3065
3066    #[test]
3067    fn test_dynamic_router_step_after_router() {
3068        // Steps after dynamic_router() are added to the outer pipeline.
3069        let definition = RouteBuilder::from("timer:tick")
3070            .route_id("test-route")
3071            .dynamic_router(Arc::new(|_| Some("mock:inner".to_string())))
3072            .to("mock:outer")
3073            .build()
3074            .unwrap();
3075
3076        assert_eq!(definition.steps().len(), 2);
3077        assert!(matches!(
3078            &definition.steps()[0],
3079            BuilderStep::DynamicRouter { .. }
3080        ));
3081        assert!(matches!(&definition.steps()[1], BuilderStep::To(uri) if uri == "mock:outer"));
3082    }
3083
3084    #[test]
3085    fn routing_slip_builder_creates_step() {
3086        use camel_api::RoutingSlipExpression;
3087
3088        let expression: RoutingSlipExpression = Arc::new(|_| Some("direct:a,direct:b".to_string()));
3089
3090        let route = RouteBuilder::from("direct:start")
3091            .route_id("routing-slip-test")
3092            .routing_slip(expression)
3093            .build()
3094            .unwrap();
3095
3096        assert!(
3097            matches!(route.steps()[0], BuilderStep::RoutingSlip { .. }),
3098            "Expected RoutingSlip step"
3099        );
3100    }
3101
3102    #[test]
3103    fn routing_slip_with_config_builder_creates_step() {
3104        use camel_api::RoutingSlipConfig;
3105
3106        let config = RoutingSlipConfig::new(Arc::new(|_| Some("mock:a".to_string())))
3107            .uri_delimiter("|")
3108            .cache_size(50)
3109            .ignore_invalid_endpoints(true);
3110
3111        let route = RouteBuilder::from("direct:start")
3112            .route_id("routing-slip-config-test")
3113            .routing_slip_with_config(config)
3114            .build()
3115            .unwrap();
3116
3117        if let BuilderStep::RoutingSlip { config } = &route.steps()[0] {
3118            assert_eq!(config.uri_delimiter, "|");
3119            assert_eq!(config.cache_size, 50);
3120            assert!(config.ignore_invalid_endpoints);
3121        } else {
3122            panic!("Expected RoutingSlip step");
3123        }
3124    }
3125
3126    #[test]
3127    fn test_builder_marshal_adds_processor_step() {
3128        let definition = RouteBuilder::from("timer:tick")
3129            .route_id("test-route")
3130            .marshal("json")
3131            .unwrap()
3132            .build()
3133            .unwrap();
3134        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3135    }
3136
3137    #[test]
3138    fn test_builder_unmarshal_adds_processor_step() {
3139        let definition = RouteBuilder::from("timer:tick")
3140            .route_id("test-route")
3141            .unmarshal("json")
3142            .unwrap()
3143            .build()
3144            .unwrap();
3145        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3146    }
3147
3148    #[test]
3149    fn test_builder_stream_cache_adds_processor_step() {
3150        let definition = RouteBuilder::from("timer:tick")
3151            .route_id("test-route")
3152            .stream_cache(1024)
3153            .build()
3154            .unwrap();
3155        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3156    }
3157
3158    #[test]
3159    fn validate_adds_validate_step() {
3160        let def = RouteBuilder::from("direct:in")
3161            .route_id("test")
3162            .validate("schemas/order.xsd")
3163            .build()
3164            .unwrap();
3165        let steps = def.steps();
3166        assert_eq!(steps.len(), 1);
3167        assert!(
3168            matches!(&steps[0], BuilderStep::Validate { predicate } if predicate.language == "simple" && predicate.source == "schemas/order.xsd"),
3169            "got: {:?}",
3170            steps[0]
3171        );
3172    }
3173
3174    #[test]
3175    fn test_builder_marshal_returns_err_for_unknown_format() {
3176        let result = RouteBuilder::from("timer:tick")
3177            .route_id("test-route")
3178            .marshal("protobuf");
3179        let err = match result {
3180            Err(e) => e,
3181            Ok(_) => panic!("marshal with unknown format should return Err"),
3182        };
3183        let msg = err.to_string();
3184        assert!(
3185            msg.contains("unknown data format"),
3186            "error should mention unknown format, got: {msg}"
3187        );
3188        assert!(
3189            msg.contains("protobuf"),
3190            "error should mention format name, got: {msg}"
3191        );
3192    }
3193
3194    #[test]
3195    fn test_builder_unmarshal_returns_err_for_unknown_format() {
3196        let result = RouteBuilder::from("timer:tick")
3197            .route_id("test-route")
3198            .unmarshal("protobuf");
3199        let err = match result {
3200            Err(e) => e,
3201            Ok(_) => panic!("unmarshal with unknown format should return Err"),
3202        };
3203        let msg = err.to_string();
3204        assert!(
3205            msg.contains("unknown data format"),
3206            "error should mention unknown format, got: {msg}"
3207        );
3208        assert!(
3209            msg.contains("protobuf"),
3210            "error should mention format name, got: {msg}"
3211        );
3212    }
3213
3214    #[test]
3215    fn test_builder_recipient_list_creates_step() {
3216        let route = RouteBuilder::from("direct:start")
3217            .route_id("recipient-list-test")
3218            .recipient_list(Arc::new(|_| "direct:a,direct:b".to_string()))
3219            .build()
3220            .unwrap();
3221
3222        assert!(matches!(
3223            &route.steps()[0],
3224            BuilderStep::RecipientList { .. }
3225        ));
3226    }
3227
3228    #[test]
3229    fn test_builder_recipient_list_with_config_creates_step() {
3230        let config = RecipientListConfig::new(Arc::new(|_| "mock:a".to_string()));
3231
3232        let route = RouteBuilder::from("direct:start")
3233            .route_id("recipient-list-config-test")
3234            .recipient_list_with_config(config)
3235            .build()
3236            .unwrap();
3237
3238        assert!(matches!(
3239            &route.steps()[0],
3240            BuilderStep::RecipientList { .. }
3241        ));
3242    }
3243
3244    #[test]
3245    fn test_builder_script_adds_script_step() {
3246        let route = RouteBuilder::from("direct:start")
3247            .route_id("script-test")
3248            .script("rhai", "headers[\"x\"] = \"y\"")
3249            .build()
3250            .unwrap();
3251
3252        assert!(matches!(
3253            &route.steps()[0],
3254            BuilderStep::Script { language, script }
3255            if language == "rhai" && script == "headers[\"x\"] = \"y\""
3256        ));
3257    }
3258
3259    #[test]
3260    fn test_builder_delay_and_delay_with_header_add_steps() {
3261        let route = RouteBuilder::from("direct:start")
3262            .route_id("delay-test")
3263            .delay(Duration::from_millis(250))
3264            .delay_with_header(Duration::from_millis(500), "x-delay")
3265            .build()
3266            .unwrap();
3267
3268        assert_eq!(route.steps().len(), 2);
3269        assert!(matches!(&route.steps()[0], BuilderStep::Delay { .. }));
3270        assert!(matches!(&route.steps()[1], BuilderStep::Delay { .. }));
3271    }
3272
3273    #[test]
3274    fn test_builder_log_and_stop_add_steps_in_order() {
3275        let route = RouteBuilder::from("direct:start")
3276            .route_id("log-stop-test")
3277            .log("hello", LogLevel::Info)
3278            .stop()
3279            .to("mock:after")
3280            .build()
3281            .unwrap();
3282
3283        assert_eq!(route.steps().len(), 3);
3284        assert!(matches!(
3285            &route.steps()[0],
3286            BuilderStep::Log { message, .. } if message == "hello"
3287        ));
3288        assert!(matches!(&route.steps()[1], BuilderStep::Stop));
3289        assert!(matches!(&route.steps()[2], BuilderStep::To(uri) if uri == "mock:after"));
3290    }
3291
3292    #[test]
3293    fn test_builder_stream_cache_default_adds_processor_step() {
3294        let route = RouteBuilder::from("direct:start")
3295            .route_id("stream-cache-default-test")
3296            .stream_cache_default()
3297            .build()
3298            .unwrap();
3299
3300        assert!(matches!(&route.steps()[0], BuilderStep::Processor(_)));
3301    }
3302
3303    #[test]
3304    fn test_validate_creates_validate_step_with_expression() {
3305        let route = RouteBuilder::from("direct:in")
3306            .route_id("validate-prefix-test")
3307            .validate("${body.size()} > 0")
3308            .build()
3309            .unwrap();
3310
3311        assert!(matches!(
3312            &route.steps()[0],
3313            BuilderStep::Validate { predicate } if predicate.language == "simple" && predicate.source == "${body.size()} > 0"
3314        ));
3315    }
3316
3317    #[test]
3318    fn test_load_balance_builder_weighted_failover_config() {
3319        let route = RouteBuilder::from("direct:start")
3320            .route_id("lb-weighted-failover")
3321            .load_balance()
3322            .weighted(vec![
3323                ("direct:a".to_string(), 3),
3324                ("direct:b".to_string(), 1),
3325            ])
3326            .failover()
3327            .to("mock:result")
3328            .end_load_balance()
3329            .build()
3330            .unwrap();
3331
3332        if let BuilderStep::LoadBalance { config, .. } = &route.steps()[0] {
3333            assert_eq!(config.strategy, LoadBalanceStrategy::Failover);
3334        } else {
3335            panic!("Expected LoadBalance step");
3336        }
3337    }
3338
3339    #[test]
3340    fn test_multicast_builder_all_config_setters() {
3341        let route = RouteBuilder::from("direct:start")
3342            .route_id("multicast-config-test")
3343            .multicast()
3344            .parallel(true)
3345            .parallel_limit(4)
3346            .stop_on_exception(true)
3347            .timeout(Duration::from_millis(300))
3348            .aggregation(MulticastStrategy::Original)
3349            .to("mock:a")
3350            .end_multicast()
3351            .build()
3352            .unwrap();
3353
3354        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3355            assert!(config.parallel);
3356            assert_eq!(config.parallel_limit, Some(4));
3357            assert!(config.stop_on_exception);
3358            assert_eq!(config.timeout, Some(Duration::from_millis(300)));
3359            assert!(matches!(config.aggregation, MulticastStrategy::Original));
3360        } else {
3361            panic!("Expected Multicast step");
3362        }
3363    }
3364
3365    #[test]
3366    fn test_build_canonical_rejects_unsupported_processor_step() {
3367        let err = RouteBuilder::from("direct:start")
3368            .route_id("canonical-reject")
3369            .set_header("k", Value::String("v".into()))
3370            .build_canonical()
3371            .unwrap_err();
3372
3373        assert!(format!("{err}").contains("does not support step `processor`"));
3374    }
3375
3376    // ── LoadBalance strategy-specific tests ─────────────────────────────────────
3377
3378    #[test]
3379    fn test_load_balance_builder_weighted_strategy() {
3380        let route = RouteBuilder::from("direct:start")
3381            .route_id("lb-weighted")
3382            .load_balance()
3383            .weighted(vec![
3384                ("direct:a".to_string(), 5),
3385                ("direct:b".to_string(), 2),
3386                ("direct:c".to_string(), 1),
3387            ])
3388            .to("mock:result")
3389            .end_load_balance()
3390            .build()
3391            .unwrap();
3392
3393        if let BuilderStep::LoadBalance { config, .. } = &route.steps()[0] {
3394            assert!(matches!(config.strategy, LoadBalanceStrategy::Weighted(_)));
3395        } else {
3396            panic!("Expected LoadBalance step");
3397        }
3398    }
3399
3400    #[test]
3401    fn test_load_balance_builder_failover_strategy() {
3402        let route = RouteBuilder::from("direct:start")
3403            .route_id("lb-failover")
3404            .load_balance()
3405            .failover()
3406            .to("mock:primary")
3407            .end_load_balance()
3408            .build()
3409            .unwrap();
3410
3411        if let BuilderStep::LoadBalance { config, .. } = &route.steps()[0] {
3412            assert_eq!(config.strategy, LoadBalanceStrategy::Failover);
3413        } else {
3414            panic!("Expected LoadBalance step");
3415        }
3416    }
3417
3418    // ── FilterInSplitBuilder tests ──────────────────────────────────────────────
3419
3420    #[test]
3421    fn test_filter_in_split_builder_typestate() {
3422        use camel_api::splitter::{SplitterConfig, split_body_lines};
3423
3424        let definition = RouteBuilder::from("timer:test")
3425            .route_id("filter-in-split")
3426            .split(SplitterConfig::new(split_body_lines()))
3427            .filter(|_ex| true)
3428            .to("mock:filtered")
3429            .end_filter()
3430            .end_split()
3431            .build()
3432            .unwrap();
3433
3434        assert_eq!(definition.steps().len(), 1);
3435        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
3436            assert_eq!(steps.len(), 1);
3437            assert!(matches!(&steps[0], BuilderStep::Filter { .. }));
3438        } else {
3439            panic!("Expected Split step");
3440        }
3441    }
3442
3443    #[test]
3444    fn test_filter_in_split_builder_multiple_steps() {
3445        use camel_api::splitter::{SplitterConfig, split_body_lines};
3446
3447        let definition = RouteBuilder::from("timer:test")
3448            .route_id("filter-in-split-multi")
3449            .split(SplitterConfig::new(split_body_lines()))
3450            .to("mock:before-filter")
3451            .filter(|_ex| true)
3452            .to("mock:inside-filter")
3453            .end_filter()
3454            .to("mock:after-filter")
3455            .end_split()
3456            .build()
3457            .unwrap();
3458
3459        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
3460            // To("before-filter") + Filter{...} + To("after-filter") = 3
3461            assert_eq!(steps.len(), 3);
3462        } else {
3463            panic!("Expected Split step");
3464        }
3465    }
3466
3467    // ── build_canonical tests ───────────────────────────────────────────────────
3468
3469    #[test]
3470    fn test_build_canonical_with_circuit_breaker() {
3471        use camel_api::circuit_breaker::CircuitBreakerConfig;
3472
3473        let spec = RouteBuilder::from("direct:start")
3474            .route_id("canonical-cb")
3475            .circuit_breaker(CircuitBreakerConfig::new().failure_threshold(10))
3476            .to("mock:result")
3477            .build_canonical()
3478            .unwrap();
3479
3480        let cb = spec.circuit_breaker.expect("circuit breaker should be set");
3481        assert_eq!(cb.failure_threshold, 10);
3482    }
3483
3484    #[test]
3485    fn test_build_canonical_rejects_custom_split_aggregation() {
3486        use camel_api::splitter::{SplitterConfig, split_body_lines};
3487
3488        let err = RouteBuilder::from("direct:start")
3489            .route_id("canonical-custom-split")
3490            .split(SplitterConfig::new(split_body_lines()).aggregation(
3491                camel_api::splitter::AggregationStrategy::Custom(Arc::new(|_, ex| ex)),
3492            ))
3493            .to("mock:frag")
3494            .end_split()
3495            .build_canonical()
3496            .unwrap_err();
3497
3498        // Split with closure-based expression is rejected in canonical v2.
3499        assert!(format!("{err}").contains("canonical v2 does not support step `split`"));
3500    }
3501
3502    #[test]
3503    fn test_build_canonical_rejects_custom_aggregate_strategy() {
3504        let err = RouteBuilder::from("direct:start")
3505            .route_id("canonical-custom-agg")
3506            .aggregate(
3507                AggregatorConfig::correlate_by("key")
3508                    .complete_when_size(2)
3509                    .strategy(AggregationStrategy::Custom(Arc::new(|_, ex| ex)))
3510                    .build()
3511                    .unwrap(),
3512            )
3513            .build_canonical()
3514            .unwrap_err();
3515
3516        assert!(format!("{err}").contains("custom aggregate strategy"));
3517    }
3518
3519    #[test]
3520    fn test_build_canonical_rejects_fn_correlation_strategy() {
3521        let err = RouteBuilder::from("direct:start")
3522            .route_id("canonical-fn-corr")
3523            .aggregate(AggregatorConfig {
3524                header_name: "key".to_string(),
3525                completion: CompletionMode::Single(CompletionCondition::Size(1)),
3526                correlation: CorrelationStrategy::Fn(Arc::new(|_| Some("key".to_string()))),
3527                strategy: AggregationStrategy::CollectAll,
3528                max_buckets: None,
3529                bucket_ttl: None,
3530                force_completion_on_stop: false,
3531                discard_on_timeout: false,
3532                max_timeout_tasks: 1024,
3533            })
3534            .build_canonical()
3535            .unwrap_err();
3536
3537        assert!(format!("{err}").contains("Fn correlation strategy"));
3538    }
3539
3540    #[test]
3541    fn test_build_canonical_rejects_predicate_completion() {
3542        let err = RouteBuilder::from("direct:start")
3543            .route_id("canonical-pred-completion")
3544            .aggregate(AggregatorConfig {
3545                header_name: "key".to_string(),
3546                completion: CompletionMode::Single(CompletionCondition::Predicate(Arc::new(
3547                    |_| false,
3548                ))),
3549                correlation: CorrelationStrategy::HeaderName("key".to_string()),
3550                strategy: AggregationStrategy::CollectAll,
3551                max_buckets: None,
3552                bucket_ttl: None,
3553                force_completion_on_stop: false,
3554                discard_on_timeout: false,
3555                max_timeout_tasks: 1024,
3556            })
3557            .build_canonical()
3558            .unwrap_err();
3559
3560        assert!(
3561            format!("{err}").contains("cannot reverse-map"),
3562            "reject message must explain forward-only: {}",
3563            err
3564        );
3565    }
3566
3567    #[test]
3568    fn extract_completion_fields_rejects_predicate_expr() {
3569        let mode = CompletionMode::Single(CompletionCondition::PredicateExpr {
3570            expr: "${body} == 'DONE'".to_string(),
3571            language: "simple".to_string(),
3572        });
3573        let result = extract_completion_fields(&mode);
3574        assert!(
3575            result.is_err(),
3576            "PredicateExpr must be rejected (forward-only)"
3577        );
3578        let msg = format!("{}", result.unwrap_err());
3579        assert!(
3580            msg.contains("cannot reverse-map"),
3581            "reject message must explain forward-only: {}",
3582            msg
3583        );
3584    }
3585
3586    #[test]
3587    fn extract_completion_fields_rejects_predicate_expr_any_mode() {
3588        let mode = CompletionMode::Any(vec![
3589            CompletionCondition::Size(5),
3590            CompletionCondition::PredicateExpr {
3591                expr: "${body} == 'DONE'".to_string(),
3592                language: "simple".to_string(),
3593            },
3594        ]);
3595        let result = extract_completion_fields(&mode);
3596        assert!(
3597            result.is_err(),
3598            "PredicateExpr in Any must be rejected (forward-only)"
3599        );
3600        let msg = format!("{}", result.unwrap_err());
3601        assert!(
3602            msg.contains("cannot reverse-map"),
3603            "reject message must explain forward-only: {}",
3604            msg
3605        );
3606    }
3607
3608    #[test]
3609    fn test_build_canonical_with_expression_correlation() {
3610        let spec = RouteBuilder::from("direct:start")
3611            .route_id("canonical-expr-corr")
3612            .aggregate(AggregatorConfig {
3613                header_name: "key".to_string(),
3614                completion: CompletionMode::Single(CompletionCondition::Size(1)),
3615                correlation: CorrelationStrategy::Expression {
3616                    expr: "header.key".to_string(),
3617                    language: "simple".to_string(),
3618                },
3619                strategy: AggregationStrategy::CollectAll,
3620                max_buckets: None,
3621                bucket_ttl: None,
3622                force_completion_on_stop: false,
3623                discard_on_timeout: false,
3624                max_timeout_tasks: 1024,
3625            })
3626            .build_canonical()
3627            .unwrap();
3628
3629        assert!(spec.steps.iter().any(|s| matches!(s, CanonicalStepSpec::Aggregate(a) if a.correlation_key == Some("header.key".to_string()))));
3630    }
3631
3632    #[test]
3633    fn test_build_canonical_split_rejected_with_closure_expression() {
3634        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
3635
3636        // Builder-based split uses closure expressions, which are not serializable.
3637        let err = RouteBuilder::from("direct:start")
3638            .route_id("canonical-split-last")
3639            .split(
3640                SplitterConfig::new(split_body_lines()).aggregation(AggregationStrategy::LastWins),
3641            )
3642            .to("mock:frag")
3643            .end_split()
3644            .build_canonical()
3645            .unwrap_err();
3646
3647        assert!(format!("{err}").contains("canonical v2 does not support step `split`"));
3648    }
3649
3650    // ── OnExceptionBuilder full chain tests ─────────────────────────────────────
3651
3652    #[test]
3653    fn test_on_exception_full_chain_retry_backoff_jitter_handled_by() {
3654        let definition = RouteBuilder::from("direct:start")
3655            .route_id("on-exception-full")
3656            .dead_letter_channel("log:dlc")
3657            .on_exception(|e| matches!(e, CamelError::Io(_)))
3658            .retry(5)
3659            .with_backoff(Duration::from_millis(10), 2.0, Duration::from_millis(500))
3660            .with_jitter(0.3)
3661            .handled_by("log:io-handler")
3662            .end_on_exception()
3663            .to("mock:out")
3664            .build()
3665            .unwrap();
3666
3667        let cfg = definition
3668            .error_handler_config()
3669            .expect("error handler should be set");
3670        assert_eq!(cfg.policies.len(), 1);
3671        let policy = &cfg.policies[0];
3672        let retry = policy.retry.as_ref().expect("retry should be set");
3673        assert_eq!(retry.max_attempts, 5);
3674        assert_eq!(retry.initial_delay, Duration::from_millis(10));
3675        assert_eq!(retry.multiplier, 2.0);
3676        assert_eq!(retry.max_delay, Duration::from_millis(500));
3677        assert!((retry.jitter_factor - 0.3).abs() < f64::EPSILON);
3678        assert_eq!(policy.handled_by.as_deref(), Some("log:io-handler"));
3679    }
3680
3681    #[test]
3682    fn test_on_exception_jitter_clamped_to_valid_range() {
3683        let definition = RouteBuilder::from("direct:start")
3684            .route_id("jitter-clamp")
3685            .on_exception(|_e| true)
3686            .retry(1)
3687            .with_jitter(5.0)
3688            .end_on_exception()
3689            .to("mock:out")
3690            .build()
3691            .unwrap();
3692
3693        let cfg = definition.error_handler_config().unwrap();
3694        let retry = cfg.policies[0].retry.as_ref().unwrap();
3695        assert!((retry.jitter_factor - 1.0).abs() < f64::EPSILON);
3696    }
3697
3698    // ── StepAccumulator: process_fn, convert_body_to, bean ──────────────────────
3699
3700    #[test]
3701    fn test_builder_process_fn_adds_processor_step() {
3702        use camel_api::BoxProcessorExt;
3703        let processor = BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }));
3704        let definition = RouteBuilder::from("timer:tick")
3705            .route_id("process-fn-test")
3706            .process_fn(processor)
3707            .build()
3708            .unwrap();
3709
3710        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3711    }
3712
3713    #[test]
3714    fn test_builder_convert_body_to_adds_processor_step() {
3715        let definition = RouteBuilder::from("timer:tick")
3716            .route_id("convert-body-test")
3717            .convert_body_to(BodyType::Json)
3718            .build()
3719            .unwrap();
3720
3721        assert!(matches!(&definition.steps()[0], BuilderStep::Processor(_)));
3722    }
3723
3724    #[test]
3725    fn test_builder_bean_adds_bean_step() {
3726        let definition = RouteBuilder::from("timer:tick")
3727            .route_id("bean-test")
3728            .bean("myBean", "process")
3729            .build()
3730            .unwrap();
3731
3732        assert!(
3733            matches!(&definition.steps()[0], BuilderStep::Bean { name, method }
3734            if name == "myBean" && method == "process")
3735        );
3736    }
3737
3738    // ── Throttle strategy-specific tests ────────────────────────────────────────
3739
3740    #[test]
3741    fn test_throttle_builder_delay_strategy() {
3742        let definition = RouteBuilder::from("timer:tick")
3743            .route_id("throttle-delay")
3744            .throttle(10, Duration::from_secs(1))
3745            .strategy(ThrottleStrategy::Delay)
3746            .to("mock:result")
3747            .end_throttle()
3748            .build()
3749            .unwrap();
3750
3751        if let BuilderStep::Throttle { config, .. } = &definition.steps()[0] {
3752            assert_eq!(config.strategy, ThrottleStrategy::Delay);
3753        } else {
3754            panic!("Expected Throttle step");
3755        }
3756    }
3757
3758    #[test]
3759    fn test_throttle_builder_drop_strategy() {
3760        let definition = RouteBuilder::from("timer:tick")
3761            .route_id("throttle-drop")
3762            .throttle(10, Duration::from_secs(1))
3763            .strategy(ThrottleStrategy::Drop)
3764            .to("mock:result")
3765            .end_throttle()
3766            .build()
3767            .unwrap();
3768
3769        if let BuilderStep::Throttle { config, .. } = &definition.steps()[0] {
3770            assert_eq!(config.strategy, ThrottleStrategy::Drop);
3771        } else {
3772            panic!("Expected Throttle step");
3773        }
3774    }
3775
3776    // ── LoopInLoopBuilder with loop_while ───────────────────────────────────────
3777
3778    #[test]
3779    fn test_nested_loop_while_builder() {
3780        use camel_api::loop_eip::LoopMode;
3781
3782        let def = RouteBuilder::from("direct:start")
3783            .route_id("nested-loop-while")
3784            .loop_count(2)
3785            .to("mock:outer")
3786            .loop_while(|_ex| true)
3787            .to("mock:inner")
3788            .end_loop()
3789            .end_loop()
3790            .build()
3791            .unwrap();
3792
3793        assert_eq!(def.steps().len(), 1);
3794        if let BuilderStep::Loop { steps, .. } = &def.steps()[0] {
3795            assert_eq!(steps.len(), 2);
3796            if let BuilderStep::Loop { config, .. } = &steps[1] {
3797                assert!(matches!(config.mode, LoopMode::While(_)));
3798            } else {
3799                panic!("Expected inner Loop step");
3800            }
3801        } else {
3802            panic!("Expected outer Loop step");
3803        }
3804    }
3805
3806    // ── Choice with multiple whens + otherwise ──────────────────────────────────
3807
3808    #[test]
3809    fn test_choice_builder_multiple_whens_with_otherwise() {
3810        let definition = RouteBuilder::from("timer:tick")
3811            .route_id("choice-multi-otherwise")
3812            .choice()
3813            .when(|ex: &Exchange| ex.input.header("a").is_some())
3814            .to("mock:a")
3815            .end_when()
3816            .when(|ex: &Exchange| ex.input.header("b").is_some())
3817            .to("mock:b")
3818            .end_when()
3819            .when(|ex: &Exchange| ex.input.header("c").is_some())
3820            .to("mock:c")
3821            .end_when()
3822            .otherwise()
3823            .to("mock:fallback")
3824            .end_otherwise()
3825            .end_choice()
3826            .build()
3827            .unwrap();
3828
3829        if let BuilderStep::Choice { whens, otherwise } = &definition.steps()[0] {
3830            assert_eq!(whens.len(), 3);
3831            assert!(otherwise.is_some());
3832            assert_eq!(otherwise.as_ref().unwrap().len(), 1);
3833        } else {
3834            panic!("Expected Choice step");
3835        }
3836    }
3837
3838    // ── Multicast individual config tests ───────────────────────────────────────
3839
3840    #[test]
3841    fn test_multicast_builder_parallel_only() {
3842        let route = RouteBuilder::from("direct:start")
3843            .route_id("multicast-parallel")
3844            .multicast()
3845            .parallel(true)
3846            .to("mock:a")
3847            .end_multicast()
3848            .build()
3849            .unwrap();
3850
3851        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3852            assert!(config.parallel);
3853            assert_eq!(config.parallel_limit, None);
3854        } else {
3855            panic!("Expected Multicast step");
3856        }
3857    }
3858
3859    #[test]
3860    fn test_multicast_builder_timeout_only() {
3861        let route = RouteBuilder::from("direct:start")
3862            .route_id("multicast-timeout")
3863            .multicast()
3864            .timeout(Duration::from_secs(5))
3865            .to("mock:a")
3866            .end_multicast()
3867            .build()
3868            .unwrap();
3869
3870        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3871            assert_eq!(config.timeout, Some(Duration::from_secs(5)));
3872        } else {
3873            panic!("Expected Multicast step");
3874        }
3875    }
3876
3877    #[test]
3878    fn test_multicast_builder_aggregation_collect_all() {
3879        let route = RouteBuilder::from("direct:start")
3880            .route_id("multicast-collect")
3881            .multicast()
3882            .aggregation(MulticastStrategy::CollectAll)
3883            .to("mock:a")
3884            .end_multicast()
3885            .build()
3886            .unwrap();
3887
3888        if let BuilderStep::Multicast { config, .. } = &route.steps()[0] {
3889            assert!(matches!(config.aggregation, MulticastStrategy::CollectAll));
3890        } else {
3891            panic!("Expected Multicast step");
3892        }
3893    }
3894
3895    // ── extract_completion_fields: Any mode with multiple conditions ────────────
3896
3897    #[test]
3898    fn test_build_canonical_aggregate_any_completion_mode() {
3899        let spec = RouteBuilder::from("direct:start")
3900            .route_id("canonical-any-completion")
3901            .aggregate(
3902                AggregatorConfig::correlate_by("key")
3903                    .complete_on_size_or_timeout(10, Duration::from_secs(30))
3904                    .build()
3905                    .unwrap(),
3906            )
3907            .build_canonical()
3908            .unwrap();
3909
3910        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3911            assert_eq!(agg.completion_size, Some(10));
3912            assert_eq!(agg.completion_timeout_ms, Some(30_000));
3913        } else {
3914            panic!("Expected Aggregate step");
3915        }
3916    }
3917
3918    #[test]
3919    fn test_build_canonical_aggregate_timeout_completion() {
3920        let spec = RouteBuilder::from("direct:start")
3921            .route_id("canonical-timeout-completion")
3922            .aggregate(
3923                AggregatorConfig::correlate_by("key")
3924                    .complete_on_timeout(Duration::from_millis(500))
3925                    .build()
3926                    .unwrap(),
3927            )
3928            .build_canonical()
3929            .unwrap();
3930
3931        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3932            assert_eq!(agg.completion_size, None);
3933            assert_eq!(agg.completion_timeout_ms, Some(500));
3934        } else {
3935            panic!("Expected Aggregate step");
3936        }
3937    }
3938
3939    // ── canonicalize_aggregate: discard_on_timeout and force_completion_on_stop ─
3940
3941    #[test]
3942    fn test_build_canonical_aggregate_discard_on_timeout() {
3943        use camel_api::aggregator::AggregatorConfig;
3944
3945        let spec = RouteBuilder::from("direct:start")
3946            .route_id("canonical-discard-timeout")
3947            .aggregate(
3948                AggregatorConfig::correlate_by("key")
3949                    .complete_when_size(1)
3950                    .discard_on_timeout(true)
3951                    .build()
3952                    .unwrap(),
3953            )
3954            .build_canonical()
3955            .unwrap();
3956
3957        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3958            assert_eq!(agg.discard_on_timeout, Some(true));
3959        } else {
3960            panic!("Expected Aggregate step");
3961        }
3962    }
3963
3964    #[test]
3965    fn test_build_canonical_aggregate_force_completion_on_stop() {
3966        use camel_api::aggregator::AggregatorConfig;
3967
3968        let spec = RouteBuilder::from("direct:start")
3969            .route_id("canonical-force-stop")
3970            .aggregate(
3971                AggregatorConfig::correlate_by("key")
3972                    .complete_when_size(1)
3973                    .force_completion_on_stop(true)
3974                    .build()
3975                    .unwrap(),
3976            )
3977            .build_canonical()
3978            .unwrap();
3979
3980        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
3981            assert_eq!(agg.force_completion_on_stop, Some(true));
3982        } else {
3983            panic!("Expected Aggregate step");
3984        }
3985    }
3986
3987    // ── build_canonical: max_buckets and bucket_ttl ─────────────────────────────
3988
3989    #[test]
3990    fn test_build_canonical_aggregate_max_buckets_and_ttl() {
3991        use camel_api::aggregator::AggregatorConfig;
3992
3993        let spec = RouteBuilder::from("direct:start")
3994            .route_id("canonical-buckets-ttl")
3995            .aggregate(
3996                AggregatorConfig::correlate_by("key")
3997                    .complete_when_size(1)
3998                    .max_buckets(100)
3999                    .bucket_ttl(Duration::from_secs(60))
4000                    .build()
4001                    .unwrap(),
4002            )
4003            .build_canonical()
4004            .unwrap();
4005
4006        if let CanonicalStepSpec::Aggregate(agg) = &spec.steps[0] {
4007            assert_eq!(agg.max_buckets, Some(100));
4008            assert_eq!(agg.bucket_ttl_ms, Some(60_000));
4009        } else {
4010            panic!("Expected Aggregate step");
4011        }
4012    }
4013
4014    // ── SplitBuilder with filter inside ─────────────────────────────────────────
4015
4016    #[test]
4017    fn test_split_builder_with_filter_inside() {
4018        use camel_api::splitter::{SplitterConfig, split_body_lines};
4019
4020        let definition = RouteBuilder::from("timer:test")
4021            .route_id("split-with-filter")
4022            .split(SplitterConfig::new(split_body_lines()))
4023            .filter(|_ex| true)
4024            .to("mock:filtered-frag")
4025            .end_filter()
4026            .end_split()
4027            .build()
4028            .unwrap();
4029
4030        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
4031            assert_eq!(steps.len(), 1);
4032            assert!(matches!(&steps[0], BuilderStep::Filter { .. }));
4033        } else {
4034            panic!("Expected Split step");
4035        }
4036    }
4037
4038    // ── WireTap additional tests ────────────────────────────────────────────────
4039
4040    #[test]
4041    fn test_wire_tap_multiple_taps() {
4042        let definition = RouteBuilder::from("timer:tick")
4043            .route_id("multi-wire-tap")
4044            .wire_tap("mock:tap1")
4045            .wire_tap("mock:tap2")
4046            .to("mock:result")
4047            .build()
4048            .unwrap();
4049
4050        assert_eq!(definition.steps().len(), 3);
4051        assert!(
4052            matches!(&definition.steps()[0], BuilderStep::WireTap { uri } if uri == "mock:tap1")
4053        );
4054        assert!(
4055            matches!(&definition.steps()[1], BuilderStep::WireTap { uri } if uri == "mock:tap2")
4056        );
4057    }
4058
4059    // ── Error handler: explicit config after shorthand → Mixed mode ─────────────
4060
4061    #[test]
4062    fn test_builder_shorthand_then_explicit_mixed_mode() {
4063        let result = RouteBuilder::from("direct:start")
4064            .route_id("mixed-mode-2")
4065            .dead_letter_channel("log:dlc")
4066            .error_handler(ErrorHandlerConfig::log_only())
4067            .to("mock:out")
4068            .build();
4069
4070        let err = result.err().expect("mixed mode should fail");
4071        assert!(format!("{err}").contains("mixed error handler modes"));
4072    }
4073
4074    // ── build_canonical: empty from_uri error ───────────────────────────────────
4075
4076    #[test]
4077    fn test_build_canonical_empty_from_uri_errors() {
4078        let result = RouteBuilder::from("").route_id("test").build_canonical();
4079        assert!(result.is_err());
4080    }
4081
4082    #[test]
4083    fn test_build_canonical_missing_route_id_errors() {
4084        let result = RouteBuilder::from("direct:start").build_canonical();
4085        assert!(result.is_err());
4086        let err = result.unwrap_err().to_string();
4087        assert!(err.contains("route_id"));
4088    }
4089
4090    // ── SplitBuilder: aggregate inside split ────────────────────────────────────
4091
4092    #[test]
4093    fn test_split_builder_with_aggregate_inside() {
4094        use camel_api::aggregator::AggregatorConfig;
4095        use camel_api::splitter::{SplitterConfig, split_body_lines};
4096
4097        let definition = RouteBuilder::from("timer:test")
4098            .route_id("split-agg")
4099            .split(SplitterConfig::new(split_body_lines()))
4100            .aggregate(
4101                AggregatorConfig::correlate_by("frag-key")
4102                    .complete_when_size(3)
4103                    .build()
4104                    .unwrap(),
4105            )
4106            .end_split()
4107            .build()
4108            .unwrap();
4109
4110        if let BuilderStep::Split { steps, .. } = &definition.steps()[0] {
4111            assert_eq!(steps.len(), 1);
4112            assert!(matches!(&steps[0], BuilderStep::Aggregate { .. }));
4113        } else {
4114            panic!("Expected Split step");
4115        }
4116    }
4117
4118    // ── Throttle: steps collected inside throttle scope ─────────────────────────
4119
4120    #[test]
4121    fn test_throttle_builder_with_steps_inside() {
4122        let definition = RouteBuilder::from("timer:tick")
4123            .route_id("throttle-steps")
4124            .throttle(10, Duration::from_secs(1))
4125            .set_header("throttled", Value::Bool(true))
4126            .to("mock:throttled")
4127            .end_throttle()
4128            .build()
4129            .unwrap();
4130
4131        if let BuilderStep::Throttle { steps, .. } = &definition.steps()[0] {
4132            assert_eq!(steps.len(), 2);
4133        } else {
4134            panic!("Expected Throttle step");
4135        }
4136    }
4137
4138    // ── LoadBalance: steps collected inside scope ───────────────────────────────
4139
4140    #[test]
4141    fn test_load_balance_builder_with_steps_inside() {
4142        let definition = RouteBuilder::from("timer:tick")
4143            .route_id("lb-steps")
4144            .load_balance()
4145            .round_robin()
4146            .set_header("lb", Value::Bool(true))
4147            .to("mock:lb")
4148            .end_load_balance()
4149            .build()
4150            .unwrap();
4151
4152        if let BuilderStep::LoadBalance { steps, .. } = &definition.steps()[0] {
4153            assert_eq!(steps.len(), 2);
4154        } else {
4155            panic!("Expected LoadBalance step");
4156        }
4157    }
4158
4159    // ── Multicast: steps collected inside scope ─────────────────────────────────
4160
4161    #[test]
4162    fn test_multicast_builder_with_steps_inside() {
4163        let definition = RouteBuilder::from("timer:tick")
4164            .route_id("multicast-steps")
4165            .multicast()
4166            .set_header("mc", Value::Bool(true))
4167            .to("mock:multicast")
4168            .end_multicast()
4169            .build()
4170            .unwrap();
4171
4172        if let BuilderStep::Multicast { steps, .. } = &definition.steps()[0] {
4173            assert_eq!(steps.len(), 2);
4174        } else {
4175            panic!("Expected Multicast step");
4176        }
4177    }
4178
4179    // ── LoopBuilder: steps collected inside loop scope ──────────────────────────
4180
4181    #[test]
4182    fn test_loop_builder_with_steps_inside() {
4183        let definition = RouteBuilder::from("timer:tick")
4184            .route_id("loop-steps")
4185            .loop_count(3)
4186            .set_header("loop", Value::Bool(true))
4187            .to("mock:loop")
4188            .end_loop()
4189            .build()
4190            .unwrap();
4191
4192        if let BuilderStep::Loop { steps, .. } = &definition.steps()[0] {
4193            assert_eq!(steps.len(), 2);
4194        } else {
4195            panic!("Expected Loop step");
4196        }
4197    }
4198
4199    // ── canonical_step_name coverage for remaining variants ─────────────────────
4200
4201    #[test]
4202    fn test_build_canonical_rejects_loop_step() {
4203        let err = RouteBuilder::from("direct:start")
4204            .route_id("canonical-loop")
4205            .loop_count(3)
4206            .to("mock:loop")
4207            .end_loop()
4208            .build_canonical()
4209            .unwrap_err();
4210
4211        assert!(format!("{err}").contains("does not support step `loop`"));
4212    }
4213
4214    #[test]
4215    fn test_build_canonical_rejects_multicast_step() {
4216        let err = RouteBuilder::from("direct:start")
4217            .route_id("canonical-multicast")
4218            .multicast()
4219            .to("mock:a")
4220            .end_multicast()
4221            .build_canonical()
4222            .unwrap_err();
4223
4224        assert!(format!("{err}").contains("does not support step `multicast`"));
4225    }
4226
4227    #[test]
4228    fn test_build_canonical_rejects_throttle_step() {
4229        let err = RouteBuilder::from("direct:start")
4230            .route_id("canonical-throttle")
4231            .throttle(10, Duration::from_secs(1))
4232            .to("mock:result")
4233            .end_throttle()
4234            .build_canonical()
4235            .unwrap_err();
4236
4237        assert!(format!("{err}").contains("does not support step `throttle`"));
4238    }
4239
4240    #[test]
4241    fn test_build_canonical_rejects_load_balancer_step() {
4242        let err = RouteBuilder::from("direct:start")
4243            .route_id("canonical-lb")
4244            .load_balance()
4245            .round_robin()
4246            .to("mock:result")
4247            .end_load_balance()
4248            .build_canonical()
4249            .unwrap_err();
4250
4251        assert!(format!("{err}").contains("does not support step `load_balancer`"));
4252    }
4253
4254    #[test]
4255    fn test_build_canonical_rejects_bean_step() {
4256        let err = RouteBuilder::from("direct:start")
4257            .route_id("canonical-bean")
4258            .bean("myBean", "process")
4259            .build_canonical()
4260            .unwrap_err();
4261
4262        assert!(format!("{err}").contains("does not support step `bean`"));
4263    }
4264
4265    #[test]
4266    fn test_build_canonical_rejects_script_step() {
4267        let err = RouteBuilder::from("direct:start")
4268            .route_id("canonical-script")
4269            .script("rhai", "x = 1")
4270            .build_canonical()
4271            .unwrap_err();
4272
4273        assert!(format!("{err}").contains("does not support step `script`"));
4274    }
4275
4276    #[test]
4277    fn test_build_canonical_accepts_delay_step() {
4278        let spec = RouteBuilder::from("direct:start")
4279            .route_id("canonical-delay")
4280            .delay(Duration::from_millis(100))
4281            .build_canonical()
4282            .unwrap();
4283
4284        assert!(
4285            spec.steps.iter().any(
4286                |s| matches!(s, CanonicalStepSpec::Delay { delay_ms, .. } if *delay_ms == 100)
4287            )
4288        );
4289    }
4290
4291    #[test]
4292    fn test_build_canonical_accepts_wire_tap_step() {
4293        let spec = RouteBuilder::from("direct:start")
4294            .route_id("canonical-wiretap")
4295            .wire_tap("mock:tap")
4296            .build_canonical()
4297            .unwrap();
4298
4299        assert!(
4300            spec.steps
4301                .iter()
4302                .any(|s| matches!(s, CanonicalStepSpec::WireTap { uri } if uri == "mock:tap"))
4303        );
4304    }
4305
4306    #[test]
4307    fn test_build_canonical_rejects_dynamic_router_step() {
4308        let err = RouteBuilder::from("direct:start")
4309            .route_id("canonical-dyn-router")
4310            .dynamic_router(Arc::new(|_| Some("mock:a".to_string())))
4311            .build_canonical()
4312            .unwrap_err();
4313
4314        assert!(format!("{err}").contains("does not support step `dynamic_router`"));
4315    }
4316
4317    #[test]
4318    fn test_build_canonical_rejects_routing_slip_step() {
4319        let err = RouteBuilder::from("direct:start")
4320            .route_id("canonical-routing-slip")
4321            .routing_slip(Arc::new(|_| Some("mock:a".to_string())))
4322            .build_canonical()
4323            .unwrap_err();
4324
4325        assert!(format!("{err}").contains("does not support step `routing_slip`"));
4326    }
4327
4328    #[test]
4329    fn test_build_canonical_rejects_recipient_list_step() {
4330        let err = RouteBuilder::from("direct:start")
4331            .route_id("canonical-recipient")
4332            .recipient_list(Arc::new(|_| "mock:a".to_string()))
4333            .build_canonical()
4334            .unwrap_err();
4335
4336        assert!(format!("{err}").contains("does not support step `recipient_list`"));
4337    }
4338
4339    // ── extract_completion_fields: Any mode with predicate → error ──────────────
4340
4341    #[test]
4342    fn test_build_canonical_rejects_any_mode_with_predicate() {
4343        let err = RouteBuilder::from("direct:start")
4344            .route_id("canonical-any-pred")
4345            .aggregate(AggregatorConfig {
4346                header_name: "key".to_string(),
4347                completion: CompletionMode::Any(vec![
4348                    CompletionCondition::Size(5),
4349                    CompletionCondition::Predicate(Arc::new(|_| false)),
4350                ]),
4351                correlation: CorrelationStrategy::HeaderName("key".to_string()),
4352                strategy: AggregationStrategy::CollectAll,
4353                max_buckets: None,
4354                bucket_ttl: None,
4355                force_completion_on_stop: false,
4356                discard_on_timeout: false,
4357                max_timeout_tasks: 1024,
4358            })
4359            .build_canonical()
4360            .unwrap_err();
4361
4362        assert!(
4363            format!("{err}").contains("cannot reverse-map"),
4364            "reject message must explain forward-only: {}",
4365            err
4366        );
4367    }
4368
4369    // ── BUILDER-004: Validation errors for missing required fields ────────────
4370
4371    #[test]
4372    fn test_builder_validation_missing_from_uri() {
4373        let result = RouteBuilder::from("")
4374            .route_id("missing-uri-route")
4375            .to("log:info")
4376            .build();
4377        assert!(result.is_err(), "empty from URI should fail validation");
4378        let err = result.err().unwrap().to_string();
4379        assert!(
4380            err.contains("'from'") || err.contains("URI"),
4381            "error should mention from/URI, got: {err}"
4382        );
4383    }
4384
4385    #[test]
4386    fn test_builder_validation_invalid_step_uri_scheme() {
4387        let result = RouteBuilder::from("timer:tick")
4388            .route_id("bad-step-route")
4389            .to("not-a-valid-uri") // no scheme
4390            .build();
4391        // The builder itself accepts any URI string; validation happens at
4392        // resolution time. Verify the build succeeds (step URI is deferred).
4393        assert!(
4394            result.is_ok(),
4395            "builder should accept opaque step URIs; resolution happens later"
4396        );
4397    }
4398
4399    // ── rc-p9vq: Duplicate route IDs ──────────────────────────────────────
4400
4401    #[test]
4402    fn test_builder_duplicate_route_ids_produce_identical_definitions() {
4403        // The builder itself doesn't check for duplicates (that's context-level).
4404        // Verify both builds succeed with the same ID — detection is TODO(rc-p9vq).
4405        let route1 = RouteBuilder::from("direct:a")
4406            .route_id("dup-route")
4407            .to("mock:out")
4408            .build();
4409        let route2 = RouteBuilder::from("direct:b")
4410            .route_id("dup-route")
4411            .to("mock:out")
4412            .build();
4413
4414        assert!(route1.is_ok());
4415        assert!(route2.is_ok());
4416        assert_eq!(route1.unwrap().route_id(), route2.unwrap().route_id());
4417    }
4418
4419    #[test]
4420    fn test_builder_clone_reuse_as_template() {
4421        // rc-8m5o: a partially-built RouteBuilder can be cloned and reused as a
4422        // template, then each clone varied independently before build().
4423        let template = RouteBuilder::from("direct:in")
4424            .set_header("stage", Value::String("shared".into()))
4425            .log("shared prefix", LogLevel::Info);
4426
4427        let route_a = template
4428            .clone()
4429            .route_id("route-a")
4430            .to("mock:a")
4431            .build()
4432            .expect("clone A builds");
4433        let route_b = template
4434            .route_id("route-b")
4435            .to("mock:b")
4436            .build()
4437            .expect("clone B builds");
4438
4439        assert_eq!(route_a.route_id(), "route-a");
4440        assert_eq!(route_b.route_id(), "route-b");
4441        // Shared template steps (set_header + log) are present in both, plus the
4442        // per-clone `to` step: the clone is a deep copy, not an alias.
4443        assert_eq!(route_a.steps().len(), route_b.steps().len());
4444        assert_eq!(route_a.from_uri(), route_b.from_uri());
4445    }
4446}
camel_builder/lib.rs

camel_builder/
lib.rs
