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

camel_builder/
lib.rs
