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

camel_builder/
lib.rs
