camel-builder

Fluent route builder API for rust-camel

Overview

camel-builder provides a fluent, type-safe API for building routes in rust-camel. Inspired by Apache Camel's Java DSL, it allows you to define integration flows using method chaining with compile-time safety.

This is the primary way to define routes in rust-camel applications.

Routes can also be defined via YAML using camel-dsl. See the camel-dsl crate for details.

Features

• Fluent API: Intuitive method chaining for route definition
• Type Safety: Compile-time verification of route structure
• All EIP Patterns: Support for filter, split, multicast, aggregate, and more
• Error Handling: Configure error handlers per route
• Circuit Breaker: Built-in resilience pattern support
• Concurrency Control: Sequential or concurrent processing modes
• Route Lifecycle: Configure auto-startup and startup order

Installation

Add to your Cargo.toml:

[dependencies]
camel-builder = "*"

Usage

Basic Route

use camel_builder::RouteBuilder;
use camel_api::Value;

let route = RouteBuilder::from("timer:tick?period=1000")
    .set_header("source", Value::String("timer".into()))
    .log("Processing tick", camel_processor::LogLevel::Info)
    .to("mock:result")
    .build()
    .unwrap();

Filter and Branching

let route = RouteBuilder::from("direct:input")
    .filter(|ex| ex.input.body.as_text().map(|t| t.starts_with("ERROR")).unwrap_or(false))
        .log("Error message detected", camel_processor::LogLevel::Warn)
        .to("direct:errors")
    .end_filter()
    .to("mock:result")
    .build()
    .unwrap();

Content-Based Router (Choice)

use camel_api::Value;

let route = RouteBuilder::from("direct:input")
    .choice()
    .when(|ex| ex.input.header("priority").map(|v| v == &Value::String("high".into())).unwrap_or(false))
        .to("direct:high-priority")
    .end_when()
    .when(|ex| ex.input.header("priority").map(|v| v == &Value::String("medium".into())).unwrap_or(false))
        .to("direct:medium-priority")
    .end_when()
    .otherwise()
        .to("direct:low-priority")
    .end_otherwise()
    .end_choice()
    .build()
    .unwrap();

Message Transformation

use camel_api::Body;

let route = RouteBuilder::from("direct:input")
    .set_header("processed", Value::Bool(true))
    .map_body(|body: Body| {
        body.as_text()
            .map(|t| Body::Text(t.to_uppercase()))
            .unwrap_or(body)
    })
    .to("mock:output")
    .build()
    .unwrap();

Splitter Pattern

use camel_api::splitter::{SplitterConfig, split_body_lines};

let route = RouteBuilder::from("file:/data/input?noop=true")
    .split(SplitterConfig::new(split_body_lines()).parallel(true))
        .log("Processing line", camel_processor::LogLevel::Debug)
        .to("direct:process-line")
    .end_split()
    .to("mock:complete")
    .build()
    .unwrap();

Multicast Pattern

let route = RouteBuilder::from("direct:input")
    .multicast()
        .parallel(true)
        .to("direct:a")
        .to("direct:b")
        .to("direct:c")
    .end_multicast()
    .to("mock:result")
    .build()
    .unwrap();

RecipientList Pattern

use camel_api::recipient_list::{RecipientListConfig, RecipientListExpression};

let expression: RecipientListExpression = Arc::new(|ex: &Exchange| {
    ex.input.header("destinations")
        .and_then(|v| v.as_str().map(String::from))
        .unwrap_or_default()
});

let route = RouteBuilder::from("direct:input")
    .recipient_list(RecipientListConfig::new(expression))
    .to("mock:result")
    .build()
    .unwrap();

Aggregate Pattern

use camel_api::aggregator::AggregatorConfig;
use std::time::Duration;

// Size-based completion (basic)
let route = RouteBuilder::from("timer:orders?period=200")
    .aggregate(
        AggregatorConfig::correlate_by("orderId")
            .complete_when_size(3)
            .build(),
    )
    .to("log:info")
    .build()
    .unwrap();

// Size OR timeout completion (v2)
let route = RouteBuilder::from("timer:orders?period=200")
    .aggregate(
        AggregatorConfig::correlate_by("orderId")
            .complete_on_size_or_timeout(3, Duration::from_secs(5))
            .force_completion_on_stop(true)
            .build(),
    )
    .to("log:info")
    .build()
    .unwrap();

Delay Pattern

use std::time::Duration;

let route = RouteBuilder::from("direct:input")
    // Delay processing by 500ms
    .delay(Duration::from_millis(500))
    // Delay with dynamic header override
    .delay_with_header(Duration::from_secs(1), "CamelDelayMs")
    .to("mock:result")
    .build()
    .unwrap();

AggregatorConfig Builder Methods

Script Step

Execute a script that can modify the Exchange in-place (headers, properties, body):

RouteBuilder::from("direct:input")
    .script("rhai", r#"headers["tenant"] = "acme"; body = body + "_processed""#)
    .to("log:result")
    .route_id("script-route")
    .build()?;

The language must be registered in the CamelContext. On error, all modifications are rolled back.

Error Handling

use camel_api::error_handler::ErrorHandlerConfig;

let route = RouteBuilder::from("direct:input")
    .error_handler(ErrorHandlerConfig::log_only())
    .process(|ex| async move {
        // Processing that might fail
        Ok(ex)
    })
    .to("mock:result")
    .build()
    .unwrap();

Shorthand route-level exception clauses:

use camel_api::CamelError;
use std::time::Duration;

let route = RouteBuilder::from("direct:input")
    .route_id("shorthand-errors")
    .dead_letter_channel("log:dlc")
    .on_exception(|e| matches!(e, CamelError::Io(_)))
        .retry(3)
        .with_backoff(Duration::from_millis(100), 2.0, Duration::from_secs(1))
        .handled_by("log:io")
    .end_on_exception()
    .on_exception(|e| matches!(e, CamelError::ProcessorError(_)))
        .retry(1)
    .end_on_exception()
    .to("mock:result")
    .build()
    .unwrap();

Circuit Breaker

use camel_api::circuit_breaker::CircuitBreakerConfig;

let route = RouteBuilder::from("direct:input")
    .circuit_breaker(CircuitBreakerConfig::new().failure_threshold(5))
    .to("http://external-service/api")
    .build()
    .unwrap();

Concurrency Control

// Concurrent processing (up to 16 in-flight)
let route = RouteBuilder::from("http://0.0.0.0:8080/api")
    .concurrent(16)
    .process(handle_request)
    .build()
    .unwrap();

// Sequential processing (ordered)
let route = RouteBuilder::from("http://0.0.0.0:8080/api")
    .sequential()
    .process(handle_request)
    .build()
    .unwrap();

Route Configuration

let route = RouteBuilder::from("timer:tick")
    .route_id("my-timer-route")
    .auto_startup(true)
    .startup_order(10)
    .to("log:info")
    .build()
    .unwrap();

Marshal/Unmarshal

use camel_builder::RouteBuilder;

let route = RouteBuilder::from("direct:in")
    .unmarshal("json")
    .marshal("xml")
    .to("mock:out")
    .route_id("marshal-unmarshal-route")
    .build()
    .unwrap();

This converts the message body from JSON to XML using the built-in data formats.

Builder Methods

Documentation

• API Documentation
• Repository

License

Apache-2.0

Contributing

Contributions are welcome! Please see the main repository for details.