Skip to main content

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