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

camel_builder/
lib.rs
