rust-camel

A Rust integration framework inspired by Apache Camel, built on Tower.

Status: Pre-release (0.1.0). APIs will change.

Overview

rust-camel lets you define message routes between components using a fluent builder API. The data plane (exchange processing, EIP patterns, middleware) is Tower-native — every processor and producer is a Service<Exchange>. The control plane (components, endpoints, consumers, lifecycle) uses its own trait hierarchy.

Current components: timer, log, direct, mock, file.

Quick Example

use camel_api::{CamelError, Value};
use camel_builder::RouteBuilder;
use camel_core::context::CamelContext;
use camel_log::LogComponent;
use camel_timer::TimerComponent;

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    tracing_subscriber::fmt::init();

    let mut ctx = CamelContext::new();

    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .set_header("source", Value::String("timer".into()))
        .to("log:info?showHeaders=true")
        .build()?;

    ctx.add_route_definition(route)?;
    ctx.start().await?;

    tokio::signal::ctrl_c()
        .await
        .map_err(|e| CamelError::Io(e.to_string()))?;

    ctx.stop().await?;
    Ok(())
}

cargo run -p hello-world

Crate Map

Building & Testing

cargo build --workspace
cargo test --workspace

Implemented EIP Patterns

Run an example:

 cargo run -p aggregator
 cargo run -p bean-demo
 cargo run -p circuit-breaker
cargo run -p content-based-routing
cargo run -p controlbus
cargo run -p error-handling
cargo run -p file-pipeline
cargo run -p file-polling
cargo run -p hello-world
cargo run -p http-client
cargo run -p http-server
cargo run -p lazy-route
cargo run -p log-eip
cargo run -p metrics-demo
cargo run -p multicast
cargo run -p multi-route-direct
cargo run -p pipeline-concurrency
cargo run -p showcase
cargo run -p splitter
cargo run -p transform-pipeline
cargo run -p wiretap

Security Features

rust-camel includes production-ready security features:

SSRF Protection (HTTP Component)

// Block private IPs by default
RouteBuilder::from("direct:start")
    .to("http://api.example.com?allowPrivateIps=false")
    .build()?

// Custom blocked hosts
RouteBuilder::from("direct:start")
    .to("http://api.example.com?blockedHosts=localhost,internal.company.com")
    .build()?

Path Traversal Protection (File Component)

All file operations validate that resolved paths remain within the configured base directory. Attempts to use ../ or absolute paths outside base are rejected.

Timeouts

All I/O operations have configurable timeouts:

• File: readTimeout, writeTimeout (default: 30s)
• HTTP: connectTimeout, responseTimeout

Memory Limits

Aggregator supports max_buckets and bucket_ttl to prevent memory leaks.

Observability

Correlation IDs

Every exchange has a unique correlation_id for distributed tracing.

Metrics

Implement MetricsCollector trait to integrate with Prometheus, OpenTelemetry, etc.

Prometheus Metrics

Export metrics to Prometheus with automatic lifecycle management:

use camel_prometheus::PrometheusService;

let ctx = CamelContext::new()
    .with_lifecycle(PrometheusService::new(9090))
    .with_tracing();

ctx.start().await?;
// Prometheus server starts automatically

Available metrics:

• camel_exchanges_total{route} - Total exchanges processed
• camel_errors_total{route, error_type} - Total errors
• camel_exchange_duration_seconds{route} - Exchange processing duration (histogram)
• camel_queue_depth{route} - Current queue depth
• camel_circuit_breaker_state{route} - Circuit breaker state

Architecture: PrometheusService implements Lifecycle trait (following Apache Camel's Service pattern, adapted to avoid tower::Service confusion).

Health Monitoring

Built-in health endpoints for Kubernetes:

• /healthz - Liveness probe
• /readyz - Readiness probe
• /health - Detailed health report

livenessProbe:
  httpGet:
    path: /healthz
    port: 9090

Route Lifecycle Management

rust-camel supports controlling when and how routes start:

Auto Startup

By default, all routes start automatically when ctx.start() is called. You can disable this:

let route = RouteBuilder::from("timer:tick")
    .route_id("lazy-route")
    .auto_startup(false)  // Won't start automatically
    .to("log:info")
    .build()?;

Startup Order

Control the order in which routes start (useful for dependencies):

let route_a = RouteBuilder::from("direct:a")
    .route_id("route-a")
    .startup_order(10)  // Starts first
    .to("log:info")
    .build()?;

let route_b = RouteBuilder::from("direct:b")
    .route_id("route-b")
    .startup_order(20)  // Starts after route-a
    .to("direct:a")
    .build()?;

Runtime Control

Control routes dynamically from code or from other routes:

// From code:
ctx.route_controller().lock().await.start_route("lazy-route").await?;
ctx.route_controller().lock().await.stop_route("route-a").await?;

// From another route (using controlbus):
RouteBuilder::from("timer:monitor")
    .set_header("CamelRouteId", Value::String("backup-route".into()))
    .to("controlbus:route?action=start")
    .build()?

See examples/lazy-route for a complete example.

Error Handling

rust-camel provides sophisticated error handling with retry policies and dead letter channels.

RedeliveryPolicy with Jitter

Configure retry behavior with exponential backoff and jitter:

use camel_api::error_handler::{ErrorHandlerConfig, RedeliveryPolicy};
use std::time::Duration;

let error_handler = ErrorHandlerConfig::dead_letter_channel("log:errors")
    .on_exception(|e| matches!(e, CamelError::Io(_)))
    .retry(3)  // Max 3 retry attempts
    .with_backoff(
        Duration::from_millis(100),  // Initial delay: 100ms
        2.0,                          // Multiplier: 2x
        Duration::from_secs(10)      // Max delay: 10s
    )
    .with_jitter(0.2)  // ±20% randomization (recommended: 0.1-0.3)
    .build();

Jitter Benefits:

• Prevents thundering herd in distributed systems
• Recommended values: 0.1-0.3 (10-30%)
• Adds randomization: delay ± (delay * jitter_factor)

Camel-Compatible Headers

During retries, these headers are automatically set:

• CamelRedelivered - true when exchange is being retried
• CamelRedeliveryCounter - Current retry attempt (1-indexed)
• CamelRedeliveryMaxCounter - Maximum retry attempts

YAML Configuration

routes:
  - id: "retry-example"
    from: "timer:tick"
    error_handler:
      dead_letter_uri: "log:dlc"
      retry:
        max_attempts: 3
        initial_delay_ms: 100
        multiplier: 2.0
        max_delay_ms: 10000
        jitter_factor: 0.2
    steps:
      - to: "direct:processor"

See examples/error-handling for complete examples.

Configuration

rust-camel supports external configuration via Camel.toml files using the camel-config crate:

Configuration File

Create a Camel.toml file:

[default]
routes = ["routes/**/*.yaml"]
log_level = "INFO"

[production]
log_level = "ERROR"

Loading Configuration

use camel_config::CamelConfig;
use camel_core::CamelContext;
use camel_timer::TimerComponent;
use camel_log::LogComponent;

// Load configuration
let config = CamelConfig::from_file("Camel.toml")?;

// Create context and register components
let mut ctx = CamelContext::new();
ctx.register_component(TimerComponent::new());
ctx.register_component(LogComponent::new());

// Load routes from discovered files
let routes = config.load_routes()?;
for route in routes {
    ctx.add_route_definition(route)?;
}

ctx.start().await?;

Route Files

Create YAML route files:

# routes/hello.yaml
routes:
  - id: "hello-timer"
    from: "timer:tick?period=1000"
    steps:
      - to: "log:info"

Environment Variables

Override configuration with environment variables:

# Select profile
export CAMEL_PROFILE=production

# Override specific values
export CAMEL_LOG_LEVEL=DEBUG
export CAMEL_ROUTES_0="custom/*.yaml"

Features

• Profile support: Multiple environments in one file
• Route discovery: Auto-load routes from glob patterns
• Environment overrides: Override any value with CAMEL_* prefix
• Deep merging: Nested configs merge properly

See docs/configuration.md for full details.

License

Apache-2.0