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

camel_builder/
lib.rs
