camel_builder/
lib.rs

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

camel_builder/
lib.rs
