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

camel_builder/
lib.rs
