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

camel_builder/
lib.rs
