Introduction

rust-camel is a Rust-native integration framework. It moves messages between timers, HTTP APIs, files, message brokers, databases, and LLMs. You compose each route from Enterprise Integration Patterns.

The pattern vocabulary comes from Apache Camel. The implementation does not. Every processor and producer is a Tower Service<Exchange>. A route is a middleware chain, so backpressure, timeouts, and cancellation are built in. The compiler checks the route before it runs. See ADR-0001 for the data-plane design.

Two interfaces build the same RouteDefinition. A fluent Rust API gives developers compile-time safety. A YAML DSL gives operators declarative authoring. Neither is a wrapper around the other.

You can build:

• HTTP APIs that fan out to Kafka and a database.
• File pipelines that watch a directory, transform, and publish.
• Message routers that split, filter, and resequence broker streams.
• Scheduled jobs that call an LLM and write the reply to a sink.

rust-camel is not a Camel port. It promises no Camel compatibility. It promises Camel familiarity. A Camel user recognizes Filter, Content-Based Router, and Splitter on first read. See ADR-0046 for the consultation protocol that governs when Camel behavior informs this design.

Start with the Getting started guide.

Getting started

Build and run your first rust-camel route. By the end, you will have a route that moves messages between two endpoints.

Prerequisites

You need the Rust toolchain (1.89 or newer). For the YAML path, also install the CLI with cargo install camel-cli. See Installation.

Choose a path

The Rust builder API gives type-safe compiled routes for application code. The YAML DSL gives declarative routes for ops-authored configuration.

1. Installation
2. First route in Rust or First route in YAML
3. CLI usage: run, scaffold, and inspect routes from the terminal.

Next, see Core concepts for the Exchange, Message, and CamelContext model.

Installation

Install rust-camel to build integration routes in Rust. Two paths: embed the crates in a Rust application, or run YAML routes with the CLI.

Prerequisites

Install the Rust toolchain with rustup. rust-camel requires Rust 1.89 or newer. The workspace pins edition 2024, and the Tokio async runtime needs a current stable release.

Confirm your toolchain:

rustc --version

Path 1: Embed in a Rust project

Add the core crates to an existing Cargo project:

cargo add camel-core camel-builder camel-api

Then add only the components your routes reference. A timer-to-log route needs two endpoint crates:

cargo add camel-component-timer camel-component-log

These commands resolve the latest published release from crates.io. The resulting [dependencies] block looks like:

[dependencies]
camel-api = "0.26"
camel-core = "0.26"
camel-builder = "0.26"
camel-component-timer = "0.26"
camel-component-log = "0.26"
tokio = { version = "1", features = ["full"] }
tracing-subscriber = "0.3"

camel-core provides CamelContext, the runtime that starts and stops routes. camel-builder provides the fluent RouteBuilder API. camel-api provides the Exchange, Message, and Value types. Each component crate is optional. Add only what a route references.

Write your first route next: First route in Rust.

Path 2: Run YAML routes with the CLI

The camel CLI parses YAML route files and starts them without compiling Rust. Install the binary from crates.io:

cargo install camel-cli

This places camel on your PATH. Confirm the install:

camel --version

Scaffold a project and run it:

camel new my-integration
cd my-integration
camel run

camel new creates a Camel.toml config file and a routes/ directory with a starter route. camel run starts the Camel context from that config.

Write your first YAML route next: First route in YAML.

Run the examples from source

The repository ships compiled examples. Clone it and run one to verify your toolchain:

git clone https://github.com/kennycallado/rust-camel.git
cd rust-camel
cargo run -p hello-world

The hello-world example fires a timer five times and logs each tick. You should see five log lines, then the process waits for Ctrl+C.

Troubleshooting

Next steps

• First route in Rust or First route in YAML
• CLI usage for every CLI command
• Core concepts for the Exchange and CamelContext model

Reference: CLI crate

First route in Rust

Build a route that produces five log messages from a timer. The route stamps a header onto each message and prints it through the log component.

The code comes from the compiled hello-world example.

The complete route

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    tracing_subscriber::fmt()
        .with_target(false) // Cleaner output
        .init();

    let mut ctx = CamelContext::builder().build().await.unwrap(); // allow-unwrap
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .route_id("hello-world") // Named routes
        .set_header("source", Value::String("timer".into()))
        .to("log:info?showHeaders=true&showCorrelationId=true") // Correlation ID
        .build()?;

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

    println!("Hello World example running. Press Ctrl+C to stop.");

    tokio::signal::ctrl_c().await.ok();
    ctx.stop().await?;
    Ok(())
}

routes:
  - id: "hello-world"
    from: "timer:tick?period=1000&repeatCount=5"
    steps:
      - set_header:
          key: "source"
          value: "timer"
      - to: "log:info?showHeaders=true&showCorrelationId=true"

Dependencies

Add these crates to your Cargo.toml:

[dependencies]
camel-api.workspace = true
camel-core.workspace = true
camel-builder.workspace = true
camel-component-timer.workspace = true
camel-component-log.workspace = true
tokio.workspace = true
tracing-subscriber.workspace = true

Each crate has one job. camel-builder gives you the fluent RouteBuilder API. camel-core gives you CamelContext, the runtime that owns routes and components. camel-component-timer and camel-component-log provide the two endpoints the route connects. camel-api provides the shared types Value and CamelError. tokio runs the async runtime. tracing-subscriber formats the log output.

How it works

Build the context

CamelContext::builder().build().await constructs the runtime. The context is the composition root for the whole process. It holds the component, language, function, and service registries. It also controls route lifecycle: start, stop, suspend, and resume. You create one context per process.

The route references two endpoint schemes, timer and log. The context resolves a scheme to a component only after you register that component. ctx.register_component(TimerComponent::new()) registers the timer scheme. ctx.register_component(LogComponent::new()) registers the log scheme. Without registration, RouteBuilder::from("timer:...") fails at build time with an unknown scheme.

Author the route

RouteBuilder::from("timer:tick?period=1000&repeatCount=5") opens a route and attaches a timer consumer. The endpoint URI has three parts. timer is the component scheme. tick is the endpoint name inside the component. The query string configures the schedule: period=1000 fires once per second, and repeatCount=5 stops the timer after five ticks.

.route_id("hello-world") names the route. Named routes are easier to inspect and to stop individually at runtime.

.set_header("source", Value::String("timer".into())) stamps a header onto every exchange. The timer consumer creates one exchange per tick. The header travels with the exchange so downstream steps can read it.

.to("log:info?showHeaders=true&showCorrelationId=true") sends each exchange to a log producer. The log component formats the exchange body and writes it through tracing. The query parameters tell the component to include the headers and the correlation ID in each output line.

.build() consumes the builder and returns a RouteDefinition. The builder is a single-shot object. You cannot clone or reuse it after build.

Register and start the route

ctx.add_route_definition(route).await hands the route to the context. The context stores the route but does not start it.

ctx.start().await starts every registered route. The timer consumer begins to fire. Each tick produces an exchange, the route stamps the header, and the log component writes a line.

tokio::signal::ctrl_c().await blocks the main task until you press Ctrl+C. ctx.stop().await then shuts the context down cleanly.

Run it

cargo run -p hello-world

The timer fires once per second. After five ticks it stops producing. The program keeps running until you press Ctrl+C.

The output shows five log lines. Each line carries the source header and a correlation ID that traces the exchange through the pipeline.

Next steps

• First route in YAML: the same route in declarative form.
• CLI usage: run, scaffold, and inspect routes from the terminal.
• Core concepts: the Exchange, Message, and CamelContext data model.

Reference: camel-builder, camel-core

First route in YAML

Declare a route in YAML that produces log messages from a timer. You write no Rust code and compile nothing. The CLI parses the YAML file and starts the route.

The YAML route and the Rust builder route lower to the same RouteDefinition. The runtime cannot tell which authoring form produced it. Pick YAML when routes live with configuration, change without a rebuild, or belong to an ops team. See ADR-0026 for the canonical authoring decision.

The route file comes from the config-basic example.

The complete route

routes:
  - id: "hello-timer"
    from: "timer:tick?period=2000&repeatCount=3"
    steps:
      - log: "Hello from config-loaded route!"
      - to: "log:info"

Project layout

A YAML route project needs two files: a route file and a Camel.toml config file.

my-integration/
├── Camel.toml          # Config: route discovery, log level
└── routes/
    └── hello.yaml      # Route definitions

Run camel new my-integration to scaffold this layout. The config file tells the CLI where to find route files. A minimal config declares the route glob and a log level:

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

The routes glob selects which files the CLI loads. The log_level sets the tracing threshold for the whole context. See CLI usage for the full config reference.

How it works

Top-level structure

routes is the top-level list. Each entry is one route definition. A file can hold many routes. Each route has three required fields: id, from, and steps.

Route identity

id is a unique name for the route. The CLI prints this id in log output and startup messages. Use it to inspect or stop one route among many at runtime.

Consumer endpoint

from is the consumer endpoint URI. timer:tick?period=2000&repeatCount=3 creates a timer that fires every two seconds, three times.

The URI has three parts. timer is the component scheme. tick is the endpoint name inside the component. The query string sets the schedule. period=2000 fires every 2000 milliseconds. repeatCount=3 stops the timer after three ticks.

Processing steps

steps is the ordered list of processing steps. The route runs them top to bottom for each exchange.

The log step writes a fixed message through tracing. The to step sends the exchange to a producer endpoint. Here to: "log:info" writes the exchange body at info level. Both step verbs map to the same processor types the Rust builder exposes.

Run it

From the project directory, run:

camel run

The CLI reads Camel.toml from the current directory. It loads every file that matches the routes glob, parses each YAML route into a RouteDefinition, and starts the context.

What you see

The timer fires three times. Each tick produces one exchange. The log step writes its message, then the to: "log:info" producer writes the exchange body. You see log output every two seconds for six seconds.

After three ticks the timer consumer stops producing. The process keeps running until you press Ctrl+C.

Next steps

• CLI usage: run, scaffold, and inspect routes from the terminal.
• YAML DSL: every step verb and route option.
• Core concepts: the Exchange, Message, and CamelContext model.

Reference: DSL crate, Config crate

CLI usage

The camel CLI runs, scaffolds, and inspects integration routes from the terminal. Install it with cargo install camel-cli.

Quick reference

camel run

Start a Camel context from YAML route files.

Trust model. camel run executes route scripts, WASM modules, and beans that the current working directory supplies. Run it only from a trusted directory.

The CLI reads Camel.toml from the current directory. The file defines route file patterns, log levels, component settings, and supervision policies.

camel run

Config file

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

The [default] profile sets routes = ["routes/*.yaml"] to discover route files. The [development] and [production] profiles override log level and watch mode.

Route file

routes:
  - id: "hello"
    from: "timer:tick?period=2000"
    steps:
      # log: messages are evaluated as Simple Language — ${...} expressions are interpolated
      - log: "Hello from camel-cli! Exchange #${header.CamelTimerCounter}"

Flags

Flag definitions live in crates/camel-cli/src/main.rs.

Expected output

The CLI starts the context, discovers routes that match the glob, and runs them. For hello.yaml above, the route logs one message every two seconds. The message text repeats with an increasing counter:

Hello from camel-cli! Exchange #1
Hello from camel-cli! Exchange #2

Press Ctrl+C (or send SIGTERM) to stop.

Profiles

Set the active profile with the CAMEL_PROFILE environment variable:

CAMEL_PROFILE=development camel run

The development profile sets log_level = "DEBUG" and watch = true. The production profile sets log_level = "WARN" and watch = false.

Hot-reload

With --watch, the CLI monitors route files for changes. The watcher groups rapid edits behind a 300 ms debounce window. Set watch_debounce_ms in Camel.toml to change it. Edits take effect without a restart.

camel run --watch

See crates/camel-config/CONTEXT.md for the debounce default.

Minimal config

A route can start without exec components or complex setup:

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

routes:
  - id: "hello"
    from: "timer:tick?period=2000"
    steps:
      - log: "Hello without exec! Exchange #${header.CamelTimerCounter}"

camel new

Scaffold a new Camel project with a Camel.toml and a routes/ directory.

camel new my-integration
cd my-integration
camel run

Layout simple writes only a [default] profile. Layout env adds [development] and [production]. Flag definitions live in crates/camel-cli/src/commands/new.rs.

Expected output

Created camel project: my-integration

Next steps:
  cd my-integration
  camel run
  camel run --watch

camel journal inspect

Read events from a redb runtime journal file. Use this command for offline debugging of a previous session.

camel journal inspect runtime.db

Flag definitions live in crates/camel-cli/src/commands/journal.rs.

Expected output

The default table format prints one row per event:

SEQ        TIMESTAMP                   EVENT                    ROUTE_ID
--------------------------------------------------------------------------------
00000001   2026-08-08T12:00:00.000Z    RouteRegistered          hello
00000002   2026-08-08T12:00:00.100Z    RouteStartRequested      hello
00000003   2026-08-08T12:00:00.250Z    RouteStarted             hello

Pass --format json to pipe events into another tool.

camel plugin

Scaffold and build WASM plugins. Plugins extend the runtime with custom processors, beans, or authorization policies. The CLI ships two subcommands: new and build.

camel plugin new

Create a plugin project from a template.

camel plugin new my-plugin

Flag definitions live in crates/camel-cli/src/commands/plugin.rs.

Expected output

Created camel processor plugin 'my-plugin'

Next steps:
  cd my-plugin
  camel plugin build

camel plugin build

Compile a plugin to the wasm32-wasip2 target and install the artifact into the project plugins directory.

camel plugin build

Run this from inside the plugin directory, or pass a path:

camel plugin build ./my-plugin

The CLI copies the compiled .wasm into the plugins directory. It reads the directory from [default.components.wasm].plugins_dir in Camel.toml. The default is plugins. See crates/camel-cli/src/commands/plugin.rs for the resolution rules.

Expected output

Built and installed plugin 'my-plugin'
  source: /path/to/my-plugin/target/wasm32-wasip2/release/my_plugin.wasm
  installed: /path/to/camel-root/plugins/my-plugin.wasm

camel openapi

Emit an OpenAPI 3.0.3 document from the rest: blocks in a YAML or JSON route file. The CLI ships one subcommand: generate.

camel openapi generate routes.yaml

Flag definitions live in crates/camel-cli/src/commands/openapi.rs.

Expected output

The command prints a pretty JSON document to stdout. The top-level openapi field is 3.0.3. Each rest: block becomes a path entry. Each operation becomes a verb under that path.

{
  "openapi": "3.0.3",
  "info": {
    "title": "Generated API",
    "version": "1.0.0"
  },
  "paths": {
    "/api/users": {
      "get": { "operationId": "listUsers" }
    }
  }
}

If a file has no rest: blocks, the command exits with an error. Validation warnings print to stderr.

Next steps

• See First route in YAML for a complete walkthrough.
• See YAML DSL for the full YAML reference.
• See Operations for health checks and monitoring.

Reference: CLI crate

OpenAPI and plugin subcommands

The CLI ships two helper families for route work outside the running context. One reads rest: blocks and emits an OpenAPI 3.0.3 document. The other scaffolds and compiles WASM plugins for the runtime.

Quick reference

camel openapi generate

Emit an OpenAPI 3.0.3 document from the rest: blocks of a YAML or JSON route file. The CLI reads the file, lowers the blocks, and prints a pretty JSON document to stdout.

camel openapi generate routes.yaml

Input file

A route file with one or more rest: blocks. Each block maps to a path entry. Each operation maps to a verb.

rest:
  - host: 0.0.0.0
    port: 9090
    path: /api/users
    operations:
      - method: GET
        operation_id: listUsers
        to: direct:listUsers
      - method: POST
        operation_id: createUser
        consumes: application/json
        produces: application/json
        success_status: 201
        to: direct:createUser
        request_schema:
          type: object
          properties:
            name:
              type: string
          required: [name]

Flags

Flag definitions live in crates/camel-cli/src/commands/openapi.rs.

Expected output

The command prints a pretty JSON document. The top-level openapi field is 3.0.3. Each rest: block becomes a path entry. Each operation becomes a verb under that path. The success response carries a description and a content schema under produces. Body verbs carry a requestBody with a content schema under consumes.

{
  "openapi": "3.0.3",
  "info": {
    "title": "Generated API",
    "version": "1.0.0"
  },
  "paths": {
    "/api/users": {
      "get": {
        "operationId": "listUsers",
        "responses": {
          "200": {
            "description": "OK",
            "content": {
              "application/json": { "schema": { "type": "object" } }
            }
          }
        }
      },
      "post": {
        "operationId": "createUser",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "type": "object",
                "properties": { "name": { "type": "string" } },
                "required": ["name"]
              }
            }
          }
        },
        "responses": {
          "201": {
            "description": "Created",
            "content": {
              "application/json": { "schema": { "type": "object" } }
            }
          }
        }
      }
    }
  }
}

How generation works

The CLI selects a parser by file extension. .yaml and .yml files go through extract_rest_blocks. .json files go through serde_json. Unknown extensions fall back to YAML parsing.

Generation fails fast on three conditions. A file with no rest: blocks exits with an error. Lowering errors fail before generation runs. Duplicate route ids fail before generation runs. Validation warnings print to stderr and do not abort the command.

A body verb with no request_schema produces a warning and a weak stub. A non-204 verb with no response schema does the same. Regenerate the document whenever the rest: blocks change. The document mirrors the runtime contract.

camel plugin new

Create a WASM plugin project from a template. The CLI writes a Cargo workspace member with the right target and the right dependencies. It also writes a sample lib.rs for the chosen plugin type.

camel plugin new my-plugin

Flag definitions live in crates/camel-cli/src/commands/plugin.rs.

Plugin types

The default is processor. Pick the type that matches the plugin role. The template files differ by type. A processor template exports a function that takes an exchange. A bean template exports a named function. An authorization-policy template exports a decision function that returns allow or deny.

Expected output

Created camel processor plugin 'my-plugin'

Next steps:
  cd my-plugin
  camel plugin build

camel plugin build

Compile a plugin to the wasm32-wasip2 target. The CLI copies the compiled .wasm into the project plugins directory.

camel plugin build

Run this from inside the plugin directory, or pass a path:

camel plugin build ./my-plugin

The build calls cargo build --target wasm32-wasip2. Install the target with rustup target add wasm32-wasip2 before the first build.

Plugins directory resolution

The CLI resolves the destination in two steps. First, it walks up from the plugin directory to find the project root. The root holds a Camel.toml or a workspace Cargo.toml.

Second, it reads [default.components.wasm].plugins_dir from Camel.toml. The default is plugins. The value must be a relative path with no .. segments. The CLI rejects absolute paths and paths that resolve outside the project root. These checks catch symlink escapes.

Expected output

Built and installed plugin 'my-plugin'
  source: /path/to/my-plugin/target/wasm32-wasip2/release/my_plugin.wasm
  installed: /path/to/camel-root/plugins/my-plugin.wasm

The CLI converts hyphens in the plugin name to underscores for the wasm artifact name. The my-super-plugin package compiles to my_super_plugin.wasm.

See also

• CLI usage for the rest of the CLI surface.
• YAML DSL for the full rest: block reference.
• WASM component for runtime plugin loading.

Reference: CLI crate | DSL crate | OpenAPI generator source | Plugin command source

Core concepts

rust-camel builds every route from three abstractions. An Exchange carries the data. A Route chains the processors that transform it. A Component owns a URI scheme and connects the route to the outside world. This section defines that model. Every pattern in the guide assumes it.

Read the pages in this order:

1. Exchange and Message - the envelope that carries the body, headers, properties, and error state. Every pattern mutates it.
2. Routes and pipelines - how a source and an ordered step list form a Tower service chain. This is where you compose patterns.
3. Components and endpoints - how a Component turns a URI into the consumers and producers that do real I/O.
4. Error handling - how faults propagate through a route, and how handlers control recovery.
5. Data plane vs control plane - why message flow and route lifecycle run on separate trait hierarchies.
6. Glossary - the canonical name for every term. Open it when a word needs a precise definition.

When the model is clear, the EIP patterns show how to compose these concepts into routing, transformation, and messaging solutions.

Exchange and Message

An Exchange is the unit of work that flows through a pipeline. It carries an input Message, an optional output Message, properties, an error state, and an exchange pattern. Each processor reads the Exchange, mutates it, and returns it. The next processor picks up the modified Exchange.

    let on_complete = RouteBuilder::from("direct:on-complete")
        .route_id("on-complete")
        .process(|mut exchange| async move {
            let body = exchange
                .input
                .body
                .as_text()
                .unwrap_or("<empty>")
                .to_string();
            exchange.input.body = Body::Text(format!("Exchange completed: {body}"));
            Ok(exchange)
        })
        .to("log:uow-complete?showBody=true&showCorrelationId=true")
        .build()?;

    let on_failure = RouteBuilder::from("direct:on-failure")
        .route_id("on-failure")
        .process(|mut exchange| async move {
            exchange.input.body = Body::Text("Exchange failed".to_string());
            Ok(exchange)
        })
        .to("log:uow-failed?showBody=true&showCorrelationId=true")
        .build()?;

    let success_route = RouteBuilder::from("timer:uow-success?delay=0&period=1200&repeatCount=4")
        .route_id("uow-success")
        .process(|mut exchange| async move {
            exchange.input.body = Body::Text("order-123".to_string());
            sleep(Duration::from_millis(450)).await;
            Ok(exchange)
        })
        .to("log:uow-main-success?showBody=true")
        .build()?
        .with_unit_of_work(UnitOfWorkConfig {
            on_complete: Some("direct:on-complete".to_string()),
            on_failure: None,
        });

    let failure_route = RouteBuilder::from("timer:uow-failure?delay=0&period=1800&repeatCount=2")
        .route_id("uow-failure")
        .process(|mut exchange| async move {
            exchange.input.body = Body::Text("will-fail".to_string());
            sleep(Duration::from_millis(300)).await;
            exchange.set_error(CamelError::ProcessorError("simulated failure".to_string()));
            Ok(exchange)
        })
        .build()?
        .with_unit_of_work(UnitOfWorkConfig {
            on_complete: None,
            on_failure: Some("direct:on-failure".to_string()),
        });

routes:
  - id: "on-complete"
    from: "direct:on-complete"
    steps:
      # The body-formatting Rust closure maps to a registered bean in YAML.
      - bean:
          name: "format-completed-body"
          method: "process"
      - to: "log:uow-complete?showBody=true&showCorrelationId=true"

  - id: "on-failure"
    from: "direct:on-failure"
    steps:
      - bean:
          name: "format-failed-body"
          method: "process"
      - to: "log:uow-failed?showBody=true&showCorrelationId=true"

  - id: "uow-success"
    from: "timer:uow-success?delay=0&period=1200&repeatCount=4"
    on_complete: "direct:on-complete"
    steps:
      - bean:
          name: "set-order-body"
          method: "process"
      - to: "log:uow-main-success?showBody=true"

  - id: "uow-failure"
    from: "timer:uow-failure?delay=0&period=1800&repeatCount=2"
    on_failure: "direct:on-failure"
    steps:
      # The Rust closure sets the body then calls set_error. A YAML route
      # needs a bean or function step to produce the failure.
      - bean:
          name: "set-body-and-fail"
          method: "process"

The include shows two routes. The uow-success route sets the body to "order-123", then forwards the Exchange to a log sink. The uow-failure route sets the body, then calls exchange.set_error(...) to mark the Exchange as failed. Both routes attach a UnitOfWorkConfig that fires a hook route when the Exchange exits the pipeline.

Exchange fields

The Exchange struct lives in crates/camel-api/src/exchange.rs. Its fields:

The pattern field defaults to InOnly. Call Exchange::new_in_out(...) to build an InOut exchange. The Consumer checks the pattern to decide whether to wait for a reply.

Message: body and headers

A Message holds the payload plus its metadata. Two fields:

• body: Body — the payload.
• headers: HashMap<String, Value> — metadata about the body. Keys are strings. Values are serde_json::Value.

Components construct the initial input Message when a Consumer fires. Processors read and write it through exchange.input and exchange.output.

Body variants

The body is a typed enum, not raw bytes. Each variant tags the payload with its format:

The Body enum is #[non_exhaustive] (ADR-0049). New variants may appear in a future release without a breaking-change semver bump.

Every variant can materialize to bytes. Call body.materialize() to collect the full payload as Bytes, or body.into_bytes(limit) to cap the read. Producers use this to serialize the body before I/O.

Typed body access

Processors read the body in two ways. For a quick string peek, call body.as_text(), which returns Option<&str>. For a typed conversion, call exchange.body_as::<T>(). This method uses the FromBody trait.

FromBody has built-in implementations for String, Vec<u8>, bytes::Bytes, and serde_json::Value. Each implementation converts from compatible body variants and rejects the rest with CamelError::TypeConversionFailed. For custom serde types, the impl_from_body_via_serde! macro generates an implementation from any type that implements DeserializeOwned.

The example shows both access styles. The on-complete hook reads the body with as_text(). It then rewrites the body with a direct field assignment: exchange.input.body = Body::Text(...).

Headers versus properties

Headers and properties are both string-keyed maps of Value. They differ in scope.

Headers live on a Message. They describe that specific message: content type, encoding, custom fields set by a component. Read a header with exchange.input.header("content-type"). Set one with exchange.input.set_header("source", "timer"). Headers belong to the message they were set on. When the pipeline swaps input for output, headers do not carry over.

Properties live on the Exchange itself. They survive body and message changes. A processor sets a property with exchange.set_property("attempt", 1). The next step reads it with exchange.property("attempt"). Use properties for cross-step state that is not part of the payload.

The error machinery uses properties. When set_error() fires, it auto-populates three property keys: CamelExceptionMessage, CamelExceptionKind, and CamelExceptionCaught. All languages read error context through these keys. Call handle_error() to set CamelExceptionHandled to true and clear the error.

Exchange through the pipeline

The Runtime wraps each Exchange in a UnitOfWorkConfig layer and passes it to the Route pipeline. Each processor receives the Exchange, transforms it, and returns it. The next processor picks up the result. This chain continues until the last step completes.

When a processor fails, it sets the error state with exchange.set_error(...). The pipeline executor detects the error and resolves the outcome. The UnitOfWorkConfig fires its on_failure hook instead of on_complete.

The pipeline resolves the Exchange to a PipelineOutcome: Completed, Stopped, or Failed (ADR-0024). See routes and pipelines for how the executor produces each outcome and how structural EIPs interact with it.

Reference: API contracts · Runtime

Routes and pipelines

A Route is a source endpoint followed by an ordered list of steps. Each step is a Processor that receives an Exchange, transforms it, and returns it. The final step forwards the result to a sink.

Route

A Route is a named message-processing pipeline. It pairs a source endpoint that emits Exchanges with a sequence of steps that transform or route them. The Runtime owns the definition.

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    tracing_subscriber::fmt()
        .with_target(false) // Cleaner output
        .init();

    let mut ctx = CamelContext::builder().build().await.unwrap(); // allow-unwrap
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .route_id("hello-world") // Named routes
        .set_header("source", Value::String("timer".into()))
        .to("log:info?showHeaders=true&showCorrelationId=true") // Correlation ID
        .build()?;

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

    println!("Hello World example running. Press Ctrl+C to stop.");

    tokio::signal::ctrl_c().await.ok();
    ctx.stop().await?;
    Ok(())
}

routes:
  - id: "hello-world"
    from: "timer:tick?period=1000&repeatCount=5"
    steps:
      - set_header:
          key: "source"
          value: "timer"
      - to: "log:info?showHeaders=true&showCorrelationId=true"

The include shows both ends of a Route. RouteBuilder::from("timer:tick?...") is the source endpoint. .to("log:info?...") is the sink. The .set_header(...) and .to(...) calls between them are the ordered steps. The source fires a new Exchange per timer tick. Each step runs in order. The sink receives the final state.

Pipeline

A Pipeline is the compiled assembly of Processors that processes an Exchange through a Route. Each Processor is a single processing unit. It can be an EIP pattern (filter, choice, split, setBody) or a custom step that receives and returns an Exchange.

The data plane runs on Tower (ADR-0001). Every processor and producer is a Service<Exchange>. The Tower Layer trait composes these services into a chain at build time. This is the architectural advantage over Apache Camel. Middleware, backpressure, timeout, and cancellation compose through one uniform trait instead of ad-hoc hooks.

Pipeline outcome and flow control

The pipeline executor produces a PipelineOutcome (ADR-0024) with three variants:

PipelineOutcome sits one layer above Tower. Tower Service<Exchange> responses stay Result<Exchange, CamelError>. A single adapter inside SequentialPipeline::call translates PipelineOutcome to Result. Completed and Stopped both become Ok. The consumer reply channel cannot distinguish them. It builds the response from the Exchange state in both cases.

Stop is successful control flow, not a failure (ADR-0024). The Exchange carries every mutation made before the Stop step. The error handler never runs for a Stop.

Structural EIPs

Steps come in two shapes. A leaf EIP (setBody, log, marshal) is one Processor that maps an Exchange to an Exchange. A structural EIP (Filter, Choice, Loop, Throttle, doTry, Split, Multicast, LoadBalance) contains child steps that form a sub-pipeline.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=6")
        .route_id("content-based-router-demo")
        .process(move |mut exchange: camel_api::Exchange| {
            let c = Arc::clone(&counter_clone);
            Box::pin(async move {
                let n = c.fetch_add(1, Ordering::SeqCst);
                let priority = match n % 3 {
                    0 => "high",
                    1 => "medium",
                    _ => "low",
                };
                exchange.input.body = Body::Text(priority.into());
                Ok(exchange)
            })
        })
        .choice()
        .when(|ex| ex.input.body.as_text() == Some("high"))
        .to("log:high-priority?showBody=true&showCorrelationId=true")
        .end_when()
        .when(|ex| ex.input.body.as_text() == Some("medium"))
        .to("log:medium-priority?showBody=true&showCorrelationId=true")
        .end_when()
        .otherwise()
        .to("log:low-priority?showBody=true&showCorrelationId=true")
        .end_otherwise()
        .end_choice()
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

routes:
  - id: "content-based-router-demo"
    from: "timer:tick?period=1000&repeatCount=6"
    error_handler:
      on_exceptions:
        - retry:
            max_attempts: 1
    steps:
      # The Rust closure that rotates the body becomes a registered bean.
      - bean:
          name: "set-rotating-priority"
          method: "process"
      - choice:
          when:
            - simple: "${body} == 'high'"
              steps:
                - to: "log:high-priority?showBody=true&showCorrelationId=true"
            - simple: "${body} == 'medium'"
              steps:
                - to: "log:medium-priority?showBody=true&showCorrelationId=true"
          otherwise:
            - to: "log:low-priority?showBody=true&showCorrelationId=true"

The include shows a Choice segment with three branches. Each .when(...) arm compiles to a sub-pipeline. The Choice evaluates predicates in order. It runs the first matching branch and skips the rest.

Structural EIPs return PipelineOutcome directly through the OutcomePipeline trait (ADR-0025). Each wraps its body as an OutcomeSegment, stored in a CompiledStep::Segment variant. This lets Stopped(ex) propagate out of the sub-pipeline with Exchange state intact. A .stop() inside any branch halts the entire route, not just the branch.

For the full processor catalog, see the Processor crate.

Reference: Runtime · Processor crate

Components and endpoints

A Component owns a URI scheme and builds the Endpoints that connect a Route to an external system. The Endpoint then creates the Consumer that pulls data in, or the Producer that sends data out.

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    tracing_subscriber::fmt()
        .with_target(false) // Cleaner output
        .init();

    let mut ctx = CamelContext::builder().build().await.unwrap(); // allow-unwrap
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .route_id("hello-world") // Named routes
        .set_header("source", Value::String("timer".into()))
        .to("log:info?showHeaders=true&showCorrelationId=true") // Correlation ID
        .build()?;

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

    println!("Hello World example running. Press Ctrl+C to stop.");

    tokio::signal::ctrl_c().await.ok();
    ctx.stop().await?;
    Ok(())
}

routes:
  - id: "hello-world"
    from: "timer:tick?period=1000&repeatCount=5"
    steps:
      - set_header:
          key: "source"
          value: "timer"
      - to: "log:info?showHeaders=true&showCorrelationId=true"

Two registrations sit before the Route. TimerComponent owns the timer: scheme. LogComponent owns the log: scheme. The Route then refers to those schemes by URI: timer:tick?period=1000&repeatCount=5 as its source, and log:info?showHeaders=true&showCorrelationId=true as its sink.

Component

A Component is a factory. It is identified by a URI scheme. Each scheme (timer, log, http, kafka) has exactly one Component. Components register into CamelContext by scheme at startup. The Runtime resolves every Route URI through this registry. The Component trait and the startup and shutdown contracts live in the Component SPI.

Endpoint

An Endpoint is an instantiated communication point. The Component creates it from a specific URI. The same Component can build many Endpoints from different URIs. timer:tick?period=1000 and timer:once?delay=2000 are two Endpoints from one TimerComponent. An Endpoint creates a Consumer for inbound traffic, or a Producer for outbound traffic.

Consumer

A Consumer is the source side of a Route. The Runtime starts it for the from: Endpoint. It is event-driven. It receives data from an external system and submits Exchanges to the Pipeline. The Consumer runs for the lifetime of the Route.

Some Components also expose a pull-based PollingConsumer. A PollingConsumer does not start a Route. The pollEnrich verb and the WASM camel_poll host function use it to read a resource on demand (ADR-0015).

Producer

A Producer is the sink side. The Runtime creates it for each to: Endpoint. It sends an Exchange to an external system. Producers are strictly write and send. Every Producer is a Tower Service<Exchange>. To read a resource mid-route, use a PollingConsumer. Do not use a producer mode for reads.

URI scheme resolution

When the Runtime builds a Route, it resolves each URI the same way:

1. Extract the scheme. This is the part before the first :. For timer:tick?period=1000, the scheme is timer.
2. Look up the registered Component for that scheme.
3. Call Component::create_endpoint(uri) to build the Endpoint.
4. The Endpoint creates the Consumer for a from: URI, or the Producer for a to: URI.

The path (tick) and the query (period=1000&repeatCount=5) belong to the Endpoint. The Component interprets them. A missing scheme or an unregistered scheme fails at startup, before the Route runs.

For the component catalog, see the Components section.

Reference: Component SPI · Components bounded context

Data plane vs control plane

rust-camel splits its runtime into two planes (ADR-0001). The data plane processes every Exchange through a Route pipeline. The control plane manages the lifecycle of Components, Endpoints, Consumers, and Routes. The split keeps the hot path fast and the cold path safe.

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    tracing_subscriber::fmt()
        .with_target(false) // Cleaner output
        .init();

    let mut ctx = CamelContext::builder().build().await.unwrap(); // allow-unwrap
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .route_id("hello-world") // Named routes
        .set_header("source", Value::String("timer".into()))
        .to("log:info?showHeaders=true&showCorrelationId=true") // Correlation ID
        .build()?;

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

    println!("Hello World example running. Press Ctrl+C to stop.");

    tokio::signal::ctrl_c().await.ok();
    ctx.stop().await?;
    Ok(())
}

routes:
  - id: "hello-world"
    from: "timer:tick?period=1000&repeatCount=5"
    steps:
      - set_header:
          key: "source"
          value: "timer"
      - to: "log:info?showHeaders=true&showCorrelationId=true"

The example shows both planes at work. The RouteBuilder::from(...) chain builds a Tower service pipeline. That pipeline is the data plane. The ctx.add_route_definition(...) and ctx.start() calls route through the control plane. Each plane has its own contracts, its own performance budget, and its own trait hierarchy.

Data plane: the hot path

The data plane processes every Exchange. Each step in the pipeline is a Tower Service<Exchange>. A Filter wraps a BoxProcessor. A Choice routes to one of several BoxProcessor arms. A WireTap forks to a secondary BoxProcessor. EIP composition maps cleanly to Tower's Service plus Layer pair (ADR-0001).

This is the hot path. Every microsecond matters. Tower's poll_ready and call protocol gives backpressure from the first step back to the Consumer. Services are cheap to clone and compose. The data plane must stay free of locks, allocations, and blocking operations. A step that blocks starves every Exchange behind it.

Control plane: the cold path

The control plane manages lifecycle. Components, Endpoints, and Consumers use their own trait hierarchy with start, stop, suspend, resume, and health operations. These operations do not fit Tower's request/response model (ADR-0001).

Lifecycle commands flow through the RuntimeBus. The bus records intent, projects route status, and starts or stops the Consumer. The control plane uses synchronous-projection CQRS with optimistic versioning and optional journal persistence (ADR-0002). Safety wins over speed here. The control plane can afford heavier abstractions because it runs on the cold path.

Why the separation exists

The split serves two goals.

Performance. The data plane must not block on control-plane locks. Every Exchange pays for the hot path. The control plane can take its time. It can persist commands, project state, and run health checks without charging that cost to throughput.

Safety. The data plane cannot mutate route state. It cannot start or stop a Consumer. It cannot change a Component registration. This isolation prevents runtime data from corrupting configuration. camel-core enforces the boundary at the module level. Each bounded context is a vertical slice with its own ports and adapters (ADR-0045). The data plane is not CQRS. The control plane is.

Trust boundary

Exchange data is untrusted (ADR-0032). Operator configuration is trusted. Headers, body, properties, and correlation keys inside an Exchange are adversary-controlled. This is the trust boundary.

No untrusted exchange datum may drive a control-plane action. It may not drive an unbounded numeric or resource decision. It may not reach an executable or interpretable sink. Every such crossing requires validation, bounding, or a capability check.

This rule prevents injection attacks. A hostile header cannot start or stop a Route. A crafted body cannot inflate a throttle limit or a loop cap. The pre-1.0 audit found eight violations of this principle. Each one became a Batch 1 security fix.

Cancellation

The control plane can stop a Route while Exchanges are in flight. A per-start tokio::task_local! cancel token carries the cancellation signal into the pipeline (ADR-0043). The step loop checks the token between steps, before each call. If the token is set, the pipeline returns Failed(ConsumerStopping).

Graceful stop drains in-flight Exchanges to completion first. The cancel check is a backstop. It fires only after the drain timeout expires, so stragglers exit at the next step boundary instead of hanging. The token is task-local, not compiled into the pipeline. A compiled-in token would survive a restart as a cancelled child and fail every new Exchange. The task-local resets on each start from a fresh child token.

Reference: camel-core crate

Error handling

Errors in rust-camel split into two mechanisms. Disposition decides what happens to an exchange after a catch matches it. The route-level error handler is the safety net that retries failed steps and routes exhausted exchanges to a dead-letter endpoint. The two compose. They do not overlap. Block-level do_try adds a third, local scope for a small group of steps.

Error model

Every failure in the data plane is a CamelError. The enum lives in crates/camel-api/src/error.rs.

CamelError is #[non_exhaustive]. New variants can arrive in any release without a breaking change. Match with a _ arm in downstream code. See crates/camel-api/CONTEXT.md.

Stop is not a CamelError. A stop step ends the route as successful control flow. The pipeline reports it as PipelineOutcome::Stopped, never as Err (ADR-0024).

How errors propagate

A pipeline runs its compiled steps in sequence. See routes and pipelines. Each step returns Result<Exchange, CamelError> to the executor. On Err, the executor calls the route error handler before the loop continues.

The handler returns a disposition. The executor translates that disposition into a PipelineOutcome. The full outcome algebra, and the adapter that maps it back to Result at the Tower boundary, live on the routes and pipelines page. This page covers only the error path.

Error disposition

Disposition is the per-catch decision. After a catch clause matches an error, the handler picks one of three dispositions. The ExceptionDisposition enum in crates/camel-api/src/error_handler.rs defines them.

Disposition runs inside the pipeline loop. It is not the retry policy and it is not the dead-letter channel. Those belong to the route-level error handler, which runs first. When the handler exhausts its retries and (optionally) sends the exchange to the dead-letter channel, it then applies the disposition. See ADR-0019.

The default disposition for a route-level onException clause is Propagate. The default for a do_try catch clause is Handled.

Route-level error handler

The route-level error_handler wraps every step in a route. It is the safety net. It does three jobs, in this order:

1. Retry the failed step with a redelivery policy.
2. Route the exchange to a dead-letter endpoint or a custom handled_by endpoint.
3. Apply the disposition from the matched onException clause.

Configure it with ErrorHandlerConfig. A global handler on CamelContext covers routes that have no per-route handler.

Dead Letter Channel

    let route1 = RouteBuilder::from("timer:route1?period=2000&repeatCount=1")
        .route_id("basic-dlc")
        .set_header("example", Value::String("basic-dlc".into()))
        .process_fn(always_fail("route1: permanent failure"))
        .error_handler(ErrorHandlerConfig::dead_letter_channel(
            "log:route1-dlc?showHeaders=true&showBody=true&showCorrelationId=true",
        ))
        .build()?;

routes:
  - id: "basic-dlc"
    from: "timer:route1?period=2000&repeatCount=1"
    error_handler:
      dead_letter_channel: "log:route1-dlc?showHeaders=true&showBody=true&showCorrelationId=true"
    steps:
      - set_header:
          key: "example"
          value: "basic-dlc"
      # The always_fail Rust closure maps to a registered bean in YAML.
      - bean:
          name: "always-fail"
          method: "process"

When a step fails and retries are absent or exhausted, the handler sends the exchange to the DLC endpoint. The exchange keeps its error state and its original message. The DLC is a fallback sink. It does not by itself stop the route. The disposition still decides whether the pipeline aborts, stops, or advances.

Retry with backoff

    let route2 = RouteBuilder::from("timer:route2?period=2000&repeatCount=1")
        .route_id("retry-backoff")
        .set_header("example", Value::String("retry-backoff".into()))
        .process_fn(fail_n_times(2))
        .error_handler(
            ErrorHandlerConfig::dead_letter_channel(
                "log:route2-dlc?showHeaders=true&showBody=true&showCorrelationId=true",
            )
            .on_exception(|_| true) // match all errors
            .retry(3)
            .with_backoff(Duration::from_millis(50), 2.0, Duration::from_secs(1))
            .build(),
        )
        .build()?;

routes:
  - id: "retry-backoff"
    from: "timer:route2?period=2000&repeatCount=1"
    error_handler:
      dead_letter_channel: "log:route2-dlc?showHeaders=true&showBody=true&showCorrelationId=true"
      on_exceptions:
        - retry:
            max_attempts: 3
            initial_delay_ms: 50
            multiplier: 2.0
            max_delay_ms: 1000
    steps:
      - set_header:
          key: "example"
          value: "retry-backoff"
      # The fail_n_times Rust closure maps to a registered bean in YAML.
      - bean:
          name: "fail-twice-then-succeed"
          method: "process"

The RedeliveryPolicy drives the retry loop:

• initial_delay sets the wait before the first retry.
• multiplier scales the wait after each attempt.
• max_delay caps the wait.
• jitter_factor randomizes the wait to avoid thundering-herd retries.

Defaults are a 100 ms initial delay, a 2x multiplier, a 10 s cap, and no jitter. See crates/camel-api/src/error_handler.rs. A retry that recovers the exchange clears the error and the pipeline advances. A retry that exhausts falls through to the DLC and then to the disposition.

OnException clauses

onException matches errors by variant or predicate. The first matching clause wins. Each clause can carry its own retry, its own handled_by endpoint, and its own disposition. Put broad clauses before specific clauses. A broad clause first shadows the rest.

Global error handler

Set a default handler on CamelContext. Routes without a per-route handler use it.

ctx.set_error_handler(ErrorHandlerConfig::dead_letter_channel(
    "log:global-dlc",
))
.await;

This is Rust API only. YAML routes compile to the same RouteDefinition but cannot express registration logic. A global handler has no YAML field. Set it on CamelContext in Rust.

Continued disposition in practice

The continued example shows the two mechanisms composing. The onException clause sets the disposition. The error handler clears the error and the pipeline advances to the next step. The DLC still receives the exchange for auditing.

    let eh_config = ErrorHandlerConfig::dead_letter_channel(
        "log:route10-dlc?showHeaders=true&showBody=true&showCorrelationId=true",
    )
    .on_exception(|e| matches!(e, CamelError::ProcessorError(_)))
    .continued(true) // ← clear error, pipeline continues to next step
    .retry(1)
    .build();

    let route10 = RouteBuilder::from("timer:route10?period=2000&repeatCount=1")
        .route_id("continued-disposition")
        .set_header("example", Value::String("continued-disposition".into()))
        .process_fn(always_fail("route10: permanent failure (continued=true)"))
        .error_handler(eh_config)
        .to("log:route10-continued?showHeaders=true&showBody=true&showCorrelationId=true")
        .build()?;

routes:
  - id: "continued-disposition"
    from: "timer:route10?period=2000&repeatCount=1"
    error_handler:
      dead_letter_channel: "log:route10-dlc?showHeaders=true&showBody=true&showCorrelationId=true"
      on_exceptions:
        - kind: "ProcessorError"
          continued: true
          retry:
            max_attempts: 1
    steps:
      - set_header:
          key: "example"
          value: "continued-disposition"
      # The always_fail Rust closure maps to a registered bean in YAML.
      - bean:
          name: "always-fail"
          method: "process"
      - to: "log:route10-continued?showHeaders=true&showBody=true&showCorrelationId=true"

Use Continued when a step is non-critical and the route should keep moving after it fails.

doTry blocks

do_try is a local error-handling scope. It wraps a group of steps in a try block with catch clauses and an optional finally clause. A handled catch does not trigger the route-level error handler. The block stays a local island. Unhandled errors bubble up to the route.

    let route1 = RouteBuilder::from("direct:catch-by-variant")
        .route_id("catch-by-variant")
        .do_try()
        .process(always_fail("boom-from-route-1"))
        .do_catch_exception(&["ProcessorError"])
        .handled()
        .process(log_marker("log:caught-by-variant"))
        .end_do_catch()
        .end_do_try()
        .build()?;

routes:
  - id: "catch-by-variant"
    from: "direct:catch-by-variant"
    steps:
      - do_try:
          steps:
            # The always_fail Rust closure maps to a registered bean in YAML.
            - bean:
                name: "always-fail"
                method: "process"
          catch:
            - exception:
                - "ProcessorError"
              disposition: "handled"
              steps:
                - bean:
                    name: "log-marker"
                    method: "process"

See the Do Try pattern page for catch-by-variant, catch-by-predicate, and finally clauses. The processor contract is in crates/camel-processor/CONTEXT.md.

doTry vs route-level error_handler

Step lifecycle and drain

Stateful steps (aggregators, idempotent repositories, resequencers) own background work that outlives a single process() call. They implement the StepLifecycle trait.

When a route stops, the runtime drains stateful steps in this order:

1. Cancel consumer intake.
2. Force-complete aggregator buckets.
3. Cancel the pipeline token.
4. Join the consumer and pipeline tasks.
5. Call shutdown(RouteStop) on each stateful step.
6. Reset the cancellation tokens.

Shutdown errors are best-effort. A failing step does not block the rest from draining. See ADR-0022.

Consumer failure supervision

A consumer that cannot continue returns an error from its task. The RuntimeBus records the route as failed. An optional supervision policy then restarts the whole route with backoff. Consumers retry transient external failures inside their normal receive loop. They do not restart their own task after a task-level failure. That decision belongs to the route control plane. See ADR-0007.

The SupervisingRouteController watches crashed routes and restarts them. Configure it with SupervisionConfig:

let ctx = CamelContext::builder()
    .supervision(SupervisionConfig {
        max_attempts: None,
        initial_delay: Duration::from_millis(500),
        backoff_multiplier: 2.0,
        max_delay: Duration::from_secs(4),
    })
    .build()
    .await?;

This is Rust API only. YAML routes compile to the same RouteDefinition but cannot express registration logic. Supervision is a CamelContext concern, not a route field.

See examples/auto-restart/ for a complete example.

Examples

• examples/error-handling/ covers the dead-letter channel, retry, onException, the continued disposition, the global handler, and the shorthand builder API.
• examples/do-try/ covers catch by variant, catch by predicate, and finally cleanup.
• examples/auto-restart/ covers the SupervisingRouteController with exponential backoff.

Glossary

Glossary of names used across this guide. Bold entries are cross-cutting domain terms registered in CONTEXT-MAP.md. Plain-text entries are foundational primitives defined in their owning crate's CONTEXT.md. The bold list is alphabetical. Each entry links its canonical guide page and the decision or crate that defines it.

Cross-cutting terms

• ArcSwap<TlsAcceptor> — atomic-swap holder for the gRPC TLS acceptor. Each accept loop loads a cert snapshot. HTTP and WS swap certs through RustlsConfig. Hot reload, ADR-0004.
• Bounded context — behavioral area of camel-core that owns its own domain vocabulary. The CQRS flavor is a per-context decision, never a crate-wide one. Architecture, ADR-0045.
• Bridged error — consumer failure that a bridge_* path converts into a synthetic error-bearing Exchange through send_and_wait. The route error handler owns the operational signal. Error handling, ADR-0012.
• CanonicalRouteSpec — versioned minimal route contract that runtime commands, config tooling, and hot-reload consume. v2 adds lifecycle metadata and rejects unsupported fields. Route structure, ADR-0011, ADR-0016.
• CircuitBreaker — DSL-declared fault tolerance pattern. It compiles into error-handling middleware, not a Pipeline Step. Circuit breaker, ADR-0019.
• ConsumerStopping — CamelError variant for producer poll_ready shutdown. It signals that the producer channel is closing. Distinct from Stop EIP. Error handling, ADR-0024.
• Credential redaction boundary — types that hold passwords, tokens, keys, or credential bytes must not expose those values through Debug or general-purpose Serialize. Use manual redaction or a tested wrapper. Auth, ADR-0051.
• Degraded — health state meaning the component can still process Exchanges: HTTP 200 on /readyz, pod Ready. Unhealthy returns HTTP 503 and marks the pod NotReady. Health.
• EnrichmentStrategy — strategy that merges the original Exchange with the polled or enriched Exchange in the enrich and pollEnrich verbs. Distinct from the EIP-22 AggregateStrategyDef family. Content enricher, ADR-0015.
• ErrorHandler — DSL declaration (ErrorHandler, OnException) that compiles into ErrorHandlerConfig and ExceptionPolicy at runtime. Error handling, ADR-0019.
• Exchange-data trust boundary — operator config is trusted. Exchange data (headers, body, properties, correlation keys) is untrusted and adversary-controlled. Such data must not reach a control-plane action, an unbounded decision, or an executable sink without validation. Planes, ADR-0032.
• ExceptionDisposition — enum (Propagate | Handled | Continued) that replaces handled: bool. Propagate returns the error upstream. Handled ends the route normally. Continued clears the error and advances. Error handling, ADR-0019.
• ForcedHealthFailure — when a Consumer crashes, HealthCheckRegistry pins the route's health entry to Unhealthy through force_unhealthy_for_route() until a ConsumerRestart replaces it with a live probe. Health.
• Handler-contract boundary — conceptual line between an error emitter and the route element that owns the failure's operational signal. Emitters inside the boundary log at warn! or below. Error handling, ADR-0012.
• LlmProvider — trait abstraction over LLM backends (OpenAI, Ollama, Mock). Camel-shaped, not siumai-shaped. All siumai imports stay in the adapter. Components, ADR-0020.
• Message — body and headers container inside an Exchange. exchange.input is the incoming Message. exchange.output is the optional reply Message. Exchange & Message.
• Module-discipline ceiling — camel-core 1.0 policy. Clean Architecture rings are enforced by module paths and boundary tests, not by crate isolation. A crate split stays a post-1.0 option. Architecture, ADR-0045.
• OpenAPI code-first generation — rest: AST compiled into an OpenAPI 3.0.3 document via camel openapi generate or camel_dsl::openapi::generate_openapi(). YAML DSL.
• OutcomePipeline — internal trait one layer above Tower for structural EIP sub-pipelines. It returns PipelineOutcome directly so Stopped(ex) keeps Exchange state intact. Error handling, ADR-0025.
• OutcomeSegment — wrapper struct over Box<dyn OutcomePipeline> with tracing and metrics hooks. It is the payload of CompiledStep::Segment. Error handling, ADR-0025.
• PipelineOutcome — enum (Completed(Exchange) | Stopped(Exchange) | Failed(CamelError)) produced by the pipeline executor one layer above Tower. Stop EIP is successful control flow, not an error. Error handling, ADR-0024.
• PollingConsumer — pull-based adapter created on demand from an Endpoint. It delivers one Exchange per call. Used by pollEnrich and the WASM camel_poll host function. Poll enrich, ADR-0015.
• ProviderMap — HashMap<String, Arc<dyn LlmProvider>> owned by LlmComponent and resolved by name from config. Not a global registry. Components, ADR-0020.
• REST DSL — declarative rest: YAML/JSON blocks that lower to http: consumer routes with JSON binding, path templates, and optional schema validation. YAML DSL.
• RetryableStep — object-safe trait that unifies BoxProcessor and OutcomeSegment for RouteErrorHandler::retry_step. One retry path serves both Tower processors and outcome-aware segments. Error handling, ADR-0019.
• Route lifecycle compensation — control-plane recovery rule. If a lifecycle side effect fails after durable intent changed, the Runtime marks the Route Failed. It reconciles the projection and publishes failure events instead of rolling history back. Routes & pipelines, ADR-0018.
• RouteChannelService — service that chains Security, CircuitBreaker (before_call), Pipeline (run_steps), and CircuitBreaker (after_result). Built only when an errorHandler is configured. Error handling, ADR-0019.
• RouteErrorHandler — trait injected into the pipeline with four async methods (match_policy, retry_step, handle_step, handle_boundary). The returned disposition drives the loop. Error handling, ADR-0019.
• SecurityPolicy — route-level authorization contract applied before normal Route Steps run. Denials return Unauthorized into route error handling. Auth, ADR-0010.
• Security defaults & fail-closed startup validation — five-disposition policy (Intent-Violation, Intent-Declaration, Require-Explicit-Choice, Safety-Primitive, Untrusted-Data-Validation) enforced by one fail-closed startup phase. Each hardened default has its own per-item flag. Auth, ADR-0033.
• ServerTlsSource — shared cert-file source struct (cert_path, key_path, client_ca_path) used by the gRPC, HTTP, and WS server components for initial TLS setup and reload. Hot reload.
• Side-effect failure — consumer failure that occurs after a successful send_and_wait, for example SQL onConsume post-processing. No route-level handler runs for it. The emitter owns the signal. Error handling, ADR-0012.
• Starting Route — externally observable Route lifecycle state between accepted start intent and confirmed Consumer or Pipeline side effect. Operators can see Starting in RouteStatusProjection. Routes & pipelines, ADR-0018.
• StopSegment — outcome-aware analog of CompiledStep::Stop for structural EIP sub-pipelines. It always returns PipelineOutcome::Stopped(ex). Stop, ADR-0024.
• Supervision — route-level crash recovery. A Consumer task failure sends a CrashNotification. The RuntimeBus records the route as Failed. An optional restart policy recreates the whole Route with backoff. Error handling, ADR-0007.
• Synchronous-projection CQRS — CQRS variant where the read-side projection updates inside the same optimistic-versioned UnitOfWork as the command. This gives strong read-model freshness with no projection lag. Planes, ADR-0002.
• System-broken error — failure that indicates corruption, a panic equivalent, or a contract violation. Always logged at error!, never downgraded. Error handling, ADR-0012.
• Template rendering language — language SPI implementation that renders templates (HTML, JSON, prompts) against Exchange data. Phase 1 covers inline templates. Phase 2 adds external file loading and hot-reload. MiniJinja, ADR-0047.
• TLS cert hot-reload — platform-wide inbound TLS certificate rotation via RuntimeCommand::ReloadTlsCerts { scheme, host, port }. The reload is idempotent and skips the journal. Hot reload, ADR-0004.
• TlsReloadHandler — trait that each TLS-terminating component implements (matches(scheme, host, port) plus async reload()). Components register it lazily in TlsReloadRegistry::global(). Hot reload.
• Vertical slice — unit of decomposition for camel-core. Each bounded context is a self-contained slice with its own domain/application/ports/adapters layout, not a shared technical layer. Architecture, ADR-0045.
• WASM sandbox capability posture — per-world grant model across Camel host functions and WASI interfaces. Camel calls use explicit scheme allowlists. WASI uses selective registration, not full-linker registration with runtime denial. Extending, ADR-0050.

Foundational primitives

Crate-local building blocks. The owning crate's CONTEXT.md is the canonical definition. These terms are intentionally not bold, so they stay out of the lint-glossary vocabulary check.

• Component — factory for Endpoints, identified by a URI scheme, registered into CamelContext. Components & endpoints, Components bounded context.
• Consumer — inbound adapter the Runtime starts for a Route's from: Endpoint. Components & endpoints, Components bounded context.
• EIP — Enterprise Integration Pattern. A Processor implemented as Tower middleware in camel-processor. EIP patterns, Processor crate.
• Endpoint — communication point a Component creates from a URI. Components & endpoints, Components bounded context.
• Exchange — data envelope carrying an input Message, an optional output Message, properties, error state, and an exchange pattern. Exchange & Message, API contracts.
• Producer — outbound adapter created for a to: Endpoint. It returns a BoxProcessor that sends an Exchange. Components & endpoints, Components bounded context.
• Processor — one processing unit in a Pipeline. The universal step contract over a Tower service. EIP patterns, API contracts.
• Route — named pipeline of a source Endpoint and an ordered step sequence. Routes & pipelines, Runtime.

Coming from Apache Camel

rust-camel shares Apache Camel's pattern vocabulary because the vocabulary is proven. The implementation is independent. The runtime, the type system, and the execution model are all different. A Camel user recognizes Filter, Content-Based Router, and Splitter. rust-camel is not a drop-in replacement and never claims Camel compatibility. See ADR-0046 for the design stance.

Core vocabulary

EIP names

Most EIP step names match between Apache Camel and rust-camel. The YAML DSL uses snake_case (wire_tap, load_balance) where Apache Camel's XML uses camelCase (wireTap, loadBalance).

Execution model

Apache Camel runs processors in a pipeline backed by a Java service architecture. rust-camel runs processors as Tower Service<Exchange> steps in a Tower middleware chain. This means:

• Every step is Clone + Send + Sync + 'static.
• Backpressure is explicit through Tower's poll_ready.
• The pipeline outcome is Completed, Stopped, or Failed (see ADR-0024). Stopped is not an error.

Error handling

See ADR-0019 for the exception disposition contract.

What rust-camel does not have

Body types

Apache Camel uses Object as the body type and relies on type converters. rust-camel uses a Body enum with five variants:

The Exchange and Message page covers typed body access. The Convert Body step changes between variants. The Marshal and Unmarshal step converts between wire formats (JSON, CSV, XML, ZIP) and these variants.

Getting started

If you are new to rust-camel, start with these pages:

EIP patterns

rust-camel implements the Enterprise Integration Patterns catalog (Hohpe and Woolf) as Tower middleware. Every pattern is a Service<Exchange>. You add it to a route builder next to a source and a sink.

The patterns group into four families:

• Routing: decide where an exchange goes next
• Transformation: change the content, format, or type of the exchange body
• Messaging: split, aggregate, reorder, and sample exchanges
• Resilience and control: protect a route from failure, limit throughput, and scope error handling

The vocabulary is shared with Apache Camel for familiarity, not compatibility (ADR-0046).

For route steps that are not EIPs (Stream Cache, Bean, Stop), see Processing steps. For the route structure that hosts these patterns, see Routes and pipelines.

Most patterns take a predicate or expression. For the available languages, see Expression languages.

Routing

Routing patterns decide where an exchange goes next. They pick a destination, branch on content, fan out to several endpoints, or distribute load. They correspond to the Message Routing category in Hohpe and Woolf.

• Message Filter — pass or drop an exchange by predicate
• Content-Based Router — route an exchange to one of several destinations by predicate
• Dynamic Router — compute the destination at runtime from exchange content
• Recipient List — broadcast an exchange to a list of endpoints computed at runtime
• Routing Slip — attach a sequence of endpoints and route through each in order
• Scatter-Gather — broadcast to a fixed list of endpoints and collect the responses
• Wire Tap — send a copy of the exchange to a side endpoint without blocking the main flow
• Multicast — send the exchange to several destinations in parallel
• Load Balancer — distribute exchanges across destination endpoints by strategy

For the route structure that hosts these patterns, see Routes and pipelines.

Message Filter

The Message Filter is a Message Router from Hohpe and Woolf. It drops exchanges that fail a predicate. A filter step evaluates a predicate on each exchange and runs its inner steps only when the predicate holds.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=10")
        .route_id("content-based-routing-demo")
        .process(move |mut exchange: camel_api::Exchange| {
            let c = Arc::clone(&counter_clone);
            Box::pin(async move {
                let n = c.fetch_add(1, Ordering::SeqCst);
                if n.is_multiple_of(2) {
                    exchange.input.body = Body::Text("important".into());
                } else {
                    exchange.input.body = Body::Text("routine".into());
                }
                Ok(exchange)
            })
        })
        .filter(|ex| ex.input.body.as_text() == Some("important"))
        .to("log:filtered?showBody=true&showCorrelationId=true")
        .end_filter()
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

- id: content-based-routing-demo
  from: timer:tick?period=1000&repeatCount=10
  error_handler:
    retry:
      max_attempts: 1
  steps:
    - filter:
        simple: "${body} == 'important'"
        steps:
          - to: log:filtered?showBody=true&showCorrelationId=true

The .filter(predicate) call takes a closure of type Fn(&Exchange) -> bool. In the included route the predicate reads ex.input.body.as_text() and keeps only exchanges whose body is the string important. The .to(...) call inside the filter scope is the inner step. It runs for exchanges the predicate accepts. The .end_filter() call closes the scope. The error_handler that follows then attaches to the route, not to the filter.

When the predicate returns false, the filter does not raise an error. It returns PipelineOutcome::Completed with the original exchange, and the inner step is skipped. The exchange then continues to any step after .end_filter(). In this route the filter is the last step before the error handler, so a filtered exchange simply ends with no log line written.

This is the rule that separates filter from choice. A filter has one branch that runs or skips. A Content-Based Router selects one branch from several. Use filter to gate a single sub-route on a yes-or-no condition. Use Content-Based Router when you must dispatch to one of many destinations.

Per ADR-0025, the filter compiles into a FilterSegment that operates on PipelineOutcome directly. The processor contract for the filter is documented in camel-processor/CONTEXT.md.

The example source is at examples/content-based-routing.

Content-Based Router

The Content-Based Router is a Message Router from Hohpe and Woolf. It inspects the exchange and routes it to one destination from a fixed set of branches.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=6")
        .route_id("content-based-router-demo")
        .process(move |mut exchange: camel_api::Exchange| {
            let c = Arc::clone(&counter_clone);
            Box::pin(async move {
                let n = c.fetch_add(1, Ordering::SeqCst);
                let priority = match n % 3 {
                    0 => "high",
                    1 => "medium",
                    _ => "low",
                };
                exchange.input.body = Body::Text(priority.into());
                Ok(exchange)
            })
        })
        .choice()
        .when(|ex| ex.input.body.as_text() == Some("high"))
        .to("log:high-priority?showBody=true&showCorrelationId=true")
        .end_when()
        .when(|ex| ex.input.body.as_text() == Some("medium"))
        .to("log:medium-priority?showBody=true&showCorrelationId=true")
        .end_when()
        .otherwise()
        .to("log:low-priority?showBody=true&showCorrelationId=true")
        .end_otherwise()
        .end_choice()
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

- id: content-based-router-demo
  from: timer:tick?period=1000&repeatCount=6
  error_handler:
    retry:
      max_attempts: 1
  steps:
    - choice:
        when:
          - simple: "${body} == 'high'"
            steps:
              - to: log:high-priority?showBody=true&showCorrelationId=true
          - simple: "${body} == 'medium'"
            steps:
              - to: log:medium-priority?showBody=true&showCorrelationId=true
        otherwise:
          - to: log:low-priority?showBody=true&showCorrelationId=true

The included route fires a timer that assigns one of three priority strings to the body. The .choice() call opens a routing block. Each .when(predicate).to(endpoint).end_when() chain defines one branch. The .otherwise().to(endpoint).end_otherwise() chain defines the fallback. The .end_choice() call closes the block so the error_handler attaches to the route, not to the choice.

The router evaluates the when predicates in order and runs the first branch that matches. It short-circuits on the first match, so it skips the remaining branches. The order of your when clauses matters when predicates overlap. Put the most specific predicate first. Each predicate is a closure of type Fn(&Exchange) -> bool. It can read the body, headers, and properties. The example reads ex.input.body.as_text() to dispatch on the string the previous process step produced.

If no predicate matches and you omit otherwise, the exchange passes through unchanged. It does not stop, and it raises no error. The route continues to the next step. Add an otherwise branch when an unmatched exchange must not proceed silently. The choice block composes with other steps. A process step inside a branch mutates the body, and the to endpoint receives that mutated body. An error handler attached after end_choice() catches errors raised inside the chosen branch.

Use the Content-Based Router when the set of destinations is fixed at build time. Use the Dynamic Router when the destination is itself data. A header value, a registry lookup, or a computation decides where the exchange goes next. CBR chooses among branches you wrote. The Dynamic Router computes an endpoint at runtime.

Per ADR-0025, the choice compiles into a ChoiceSegment that returns PipelineOutcome directly, so Stop propagates with exchange state intact. Per ADR-0001, that segment runs as a Service<Exchange> in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

For how a filter feeds a choice inside the same route, see the Message Filter page.

The example source is at examples/content-based-router.

Dynamic Router

The Dynamic Router is a Message Router from Hohpe and Woolf. It computes the destination endpoint at runtime from exchange data, instead of selecting from a fixed set of branches.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=10")
        .route_id("dynamic-router-demo")
        // Set a rotating destination header
        .process(move |mut exchange: camel_api::Exchange| {
            let dests = dest_clone.clone();
            Box::pin(async move {
                let mut d = dests.lock().unwrap(); // allow-unwrap
                let dest = d.next().unwrap(); // allow-unwrap
                exchange
                    .input
                    .set_header("destination", Value::String(dest.to_string()));
                exchange.input.body = Body::Text(format!("routed to {}", dest));
                Ok(exchange)
            })
        })
        // Dynamic router: read header and route to correct endpoint
        .dynamic_router(Arc::new(|exchange: &camel_api::Exchange| {
            let dest = exchange
                .input
                .header("destination")
                .and_then(|v| v.as_str())
                .unwrap_or("a");
            Some(format!(
                "log:routed-{}?showBody=true&showHeaders=true",
                dest
            ))
        }))
        .build()?;

- id: dynamic-router-demo
  from: timer:tick?period=1000&repeatCount=10
  steps:
    - set_header:
        key: destination
        value: a
    - dynamic_router:
        simple: "log:routed-${header.destination}?showBody=true&showHeaders=true"

The .dynamic_router(Arc::new(|exchange| ...)) step takes a closure of type Fn(&Exchange) -> Option<String>. The router calls the closure and forwards the exchange to the endpoint it returns. Then it calls the closure again on the result. The loop ends when the closure returns None. Each hop receives the exchange as the previous endpoint left it. That endpoint can mutate a header or the body. The next closure call then reads the changed value and either returns a new destination or None to stop.

The example reads the destination header that an upstream process step sets. It returns the matching log:routed-{dest} endpoint. A single Some value may also carry several endpoints separated by the uri_delimiter (default ,). The router visits all of them within one iteration before it calls the closure again.

Two safeguards stop a runaway loop. The closure must not return the same endpoint on consecutive iterations. A hop that leaves the routing data unchanged trips this guard. The router then raises an error instead of spinning. An iteration cap (max_iterations, default 1000) and an optional timeout bound the loop. To end routing after a single hop, have that hop clear the value the closure reads, or return None.

The Dynamic Router differs from the Recipient List. The Dynamic Router re-evaluates its expression after every hop, so each endpoint can steer the exchange to the next. The Recipient List evaluates its expression once and sends the exchange to every endpoint on that list. Use the Dynamic Router when each hop can change where the exchange goes next. Use the Recipient List for one-shot fan-out to a known set.

The closure runs synchronously on the route channel. Any I/O the closure performs blocks the next step until the closure returns. If you need async work, resolve the destination in an upstream process step and store it in a header. Let the dynamic router read that header. This keeps the router a small, fast step.

Per ADR-0001, the dynamic router compiles into a DynamicRouterService that runs as a Service<Exchange> in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/dynamic-router.

Recipient List

The Recipient List is a Message Router from Hohpe and Woolf. It evaluates an expression once to resolve a list of endpoints, then sends a copy of the exchange to each one.

    let route = RouteBuilder::from("timer:tick?period=2000&repeatCount=3")
        .route_id("recipientlist-demo")
        .set_header(
            "destinations",
            Value::String("log:channel-a?showBody=true,log:channel-b?showBody=true,log:channel-c?showBody=true".into()),
        )
        .recipient_list_with_config(
            RecipientListConfig::new(expression)
                .parallel(true),
        )
        .to("log:summary?showBody=true")
        .build()?;

- id: recipientlist-demo
  from: timer:tick?period=2000&repeatCount=3
  steps:
    - set_header:
        key: destinations
        value: "log:channel-a?showBody=true,log:channel-b?showBody=true,log:channel-c?showBody=true"
    - recipient_list:
        simple: "${header.destinations}"
        parallel: true
    - to: log:summary?showBody=true

The route stores three log endpoint URIs in the destinations header. The .recipient_list_with_config(...) call takes a closure of type RecipientListExpression. That closure reads the header and returns the comma-separated string. The processor splits the string, resolves each URI, and dispatches a clone of the exchange to each endpoint. The .parallel(true) flag runs the three dispatches concurrently instead of one after the other.

The exchange that reaches the next step depends on the aggregation strategy. The default LastWins strategy forwards one branch's result to the step that follows. Set MulticastStrategy::Original to pass the input exchange through unchanged and discard every branch output. Set CollectAll to gather each branch body into a JSON array. The example keeps the default, so log:summary receives the result of one resolved branch.

Use the Recipient List when the destinations are data. A header, a registry lookup, or a computation decides the targets at runtime. Use Multicast when the targets are fixed in the route. A Content-Based Router picks one branch from a set. The Recipient List fans out to all of them. A guard caps the resolved list at max_recipients (default 1000) before any endpoint resolves. An expression that returns millions of URIs cannot exhaust memory.

Per ADR-0025, the recipient list is an outcome-aware structural EIP. A branch that returns PipelineOutcome::Stopped records a partial failure. The aggregation skips the stopped branch, and the route error handler sees the partial outcome through the same boundary that step errors flow through.

Per ADR-0001, the recipient list compiles into a Service<Exchange> step in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/recipientlist.

Routing Slip

The Routing Slip is a Message Router from Hohpe and Woolf. It reads a header that holds a comma-separated list of endpoints. It routes the exchange through each one in sequence.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=10")
        .route_id("routing-slip-demo")
        .process(move |mut exchange: camel_api::Exchange| {
            let c = counter_clone.clone();
            Box::pin(async move {
                let n = c.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let slip = if n.is_multiple_of(2) {
                    "log:step-a?showBody=true,log:step-b?showBody=true"
                } else {
                    "log:step-c?showBody=true,log:step-d?showBody=true"
                };
                exchange
                    .input
                    .set_header("slip", Value::String(slip.to_string()));
                exchange.input.body = Body::Text(format!("message #{}", n));
                Ok(exchange)
            })
        })
        .routing_slip(Arc::new(|exchange: &camel_api::Exchange| {
            exchange
                .input
                .header("slip")
                .and_then(|v| v.as_str().map(|s| s.to_string()))
        }))
        .build()?;

- id: routing-slip-demo
  from: timer:tick?period=1000&repeatCount=10
  steps:
    - set_header:
        key: slip
        value: "log:step-a?showBody=true,log:step-b?showBody=true"
    - routing_slip:
        simple: "${header.slip}"

The route sets a slip header that holds one of two endpoint lists, then calls .routing_slip(...). The closure takes a &Exchange and returns an Option<String>. It reads the slip header and returns the string. The processor splits the string on the uri_delimiter (default ,), resolves each URI, and calls each endpoint in turn. The exchange that one endpoint returns feeds into the next. If the closure returns None, the exchange passes through unchanged.

Each step sees the mutations the previous step made. A log endpoint prints the body. An endpoint can append a header or transform the body. Any of these changes what the next endpoint receives. In the example the body carries the identifier message #N, and that identifier propagates from the first step to the last.

The slip is computed once. The closure runs on the initial exchange, and the resulting list drives the whole sequence. Each endpoint mutates the exchange, but the list of endpoints itself is fixed. To change the path between exchanges, vary the header value upstream of the slip, as the example does with its counter.

The Routing Slip is sequential. The Recipient List also evaluates its expression once. It sends a copy to every endpoint and aggregates the results, instead of threading one exchange through a chain. Use the Routing Slip when the exchange must visit a sequence of endpoints in order, each one building on the last. Use the Recipient List for one-shot fan-out to an independent set. The Dynamic Router re-evaluates its destination after every hop, so it suits a path that each endpoint steers.

Per ADR-0001, the routing slip compiles into a RoutingSlipService that runs as a Service<Exchange> in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/routing-slip.

Scatter-Gather

The Scatter-Gather is a Message Router from Hohpe and Woolf. It broadcasts a copy of the exchange to a fixed list of endpoints in parallel and gathers the responses into one aggregate exchange.

- id: scatter-gather-demo
  from: timer:tick?period=1000&repeatCount=3
  steps:
    - scatter_gather:
        endpoints:
          - direct:pricing
          - direct:inventory
          - direct:reviews
        aggregation: collect_all
    - to: log:aggregated?showBody=true

The scatter_gather step sends the same exchange to every endpoint listed under endpoints. All dispatches run in parallel. The aggregation field picks the strategy that merges the responses. The default is last_wins, which keeps the body of the last branch to complete. Set aggregation: collect_all to assemble every branch body into a JSON array. The step after scatter_gather receives the merged result.

Scatter-Gather is DSL sugar over Multicast. The YAML parser lowers the step to a multicast block with parallel: true and the chosen aggregation. No new processor or Rust builder method exists for it. Rust code calls .multicast() directly. Use Scatter-Gather when the broadcast-and-collect shape is the point. Use a raw multicast block when you need the extra knobs it exposes: parallel_limit, stop_on_exception, or timeout_ms.

The endpoints are fixed in the route definition. A route that must compute destinations at runtime uses a Recipient List. The gather is also stateless. No correlation key or completion condition exists. The step collects all parallel responses in one pass and moves on. Stateful accumulation across many exchanges is the Aggregator EIP.

Per ADR-0025, the lowered multicast compiles into a MulticastSegment that operates on PipelineOutcome. Per ADR-0001, the segment runs as a Service<Exchange> step in the Tower pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

Wire Tap

The Wire Tap is a Message Router from Hohpe and Woolf. It sends a copy of the exchange to a tap endpoint for inspection or monitoring while the original exchange continues down the route unchanged.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .route_id("wiretap-demo")
        // Tap: send a clone to monitoring (fire-and-forget)
        .wire_tap("log:monitor?showBody=true&showCorrelationId=true")
        // Main pipeline continues normally
        .to("log:main?showBody=true&showCorrelationId=true")
        .build()?;

- id: wiretap-demo
  from: timer:tick?period=1000&repeatCount=5
  steps:
    - wire_tap: log:monitor?showBody=true&showCorrelationId=true
    - to: log:main?showBody=true&showCorrelationId=true

The included route fires a timer and calls .wire_tap("log:monitor?..."). The tap clones the exchange and dispatches the clone to log:monitor, which logs the body and correlation id. The main pipeline then runs the original exchange through .to("log:main?..."), which logs the same body. The tap runs as fire-and-forget. An error on the tap endpoint does not stop the main flow, and the main flow does not wait for the tap to finish.

The Wire Tap differs from Multicast in scope. A wire tap is one side channel. It does not return a value into the main pipeline. The exchange that continues down the route is the same exchange the tap saw before the clone. Multicast is the main flow: it fans out to every branch and aggregates the results back into the pipeline. Use a wire tap to observe an exchange without changing it. Put a consumer on the main pipeline when it must affect the result.

The clone duplicates the body. A wire tap on an exchange with a multi-megabyte body copies that body for the tap endpoint, which costs memory and CPU. For large bodies, pair the route with a Claim Check step that stashes the body and passes a reference id. The tap then reads the reference id without copying the body.

Per ADR-0001, the wire tap compiles into a Service<Exchange> step in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/wiretap.

Multicast

The Multicast is a Message Router from Hohpe and Woolf. It sends a copy of the exchange to every endpoint in a fixed list and merges the responses back into one exchange.

    let route = RouteBuilder::from("timer:tick?period=2000&repeatCount=3")
        .route_id("multicast-demo")
        // Give the exchange a meaningful body before multicasting
        .process(|mut exchange: camel_api::Exchange| {
            Box::pin(async move {
                exchange.input.body = Body::Text("hello from multicast".to_string());
                // Add broadcast-id header with UUID
                exchange
                    .input
                    .set_header("broadcast-id", Value::String(Uuid::new_v4().to_string()));
                Ok(exchange)
            })
        })
        // Multicast: same message goes to all three log endpoints in parallel.
        // CollectAll aggregation gathers each response body into a JSON array.
        .multicast()
        .parallel(true)
        .aggregation(MulticastStrategy::CollectAll)
        .to("log:channel-a?showBody=true&showCorrelationId=true")
        .to("log:channel-b?showBody=true&showCorrelationId=true")
        .to("log:channel-c?showBody=true&showCorrelationId=true")
        .end_multicast()
        // After multicast: the body is a JSON array of all three responses
        .to("log:summary?showBody=true&showCorrelationId=true")
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

- id: multicast-demo
  from: timer:tick?period=2000&repeatCount=3
  error_handler:
    retry:
      max_attempts: 1
  steps:
    - set_body:
        value: hello from multicast
    - multicast:
        parallel: true
        aggregation: collect_all
        steps:
          - to: log:channel-a?showBody=true&showCorrelationId=true
          - to: log:channel-b?showBody=true&showCorrelationId=true
          - to: log:channel-c?showBody=true&showCorrelationId=true
    - to: log:summary?showBody=true&showCorrelationId=true

The example sets a text body and a broadcast-id header, then opens a multicast block over three log endpoints. The parallel: true flag dispatches all three branches at once instead of one after another. Each branch receives its own clone of the exchange, so a branch that mutates the body or headers does not affect its siblings. The clone also carries a CamelMulticastIndex property and a CamelMulticastComplete flag. A branch can read these to learn its position in the fan-out. The collect_all strategy gathers each branch response body into a JSON array, and the step that follows the block, log:summary, receives that array as its body.

The aggregation strategy decides what the post-multicast exchange carries. LastWins (the default) keeps the body of the last branch to complete and discards the rest. CollectAll assembles every branch body into a JSON array. Original passes the input exchange through unchanged and drops all branch output. A Custom variant takes a function of type MulticastAggregationFn for merge logic the built-ins do not cover. Three other knobs live on the block. parallel_limit caps how many branches run at once when parallel is on. stop_on_exception fails the whole block on the first branch error. timeout bounds how long the block waits for slow branches.

Pick Multicast when the destinations are fixed in the route and you want every one to run. Pick the Recipient List when a header or expression must compute the destinations at runtime. Scatter-Gather is YAML sugar over Multicast with parallel dispatch and collect_all. Use it when the broadcast-and-collect shape is the whole point. A Wire Tap sends one copy to a side channel and never merges a result back.

Per ADR-0025, multicast is an outcome-aware structural EIP. A branch that returns PipelineOutcome::Stopped propagates that stop out of the block. Aggregation does not run on stopped output. The route error handler sees any partial outcome through the same boundary that step errors use. Per ADR-0001, the block compiles into a Service<Exchange> step in the Tower pipeline, with each branch a child step on the route channel. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/multicast.

Load Balancer

The Load Balancer is a Message Router from Hohpe and Woolf. For each exchange it picks one endpoint from a fixed list and sends the exchange there.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=10")
        .route_id("load-balancer-demo")
        // Give the exchange a meaningful body before load balancing
        .process(|mut exchange: camel_api::Exchange| {
            Box::pin(async move {
                exchange.input.body = Body::Text("hello from load balancer".to_string());
                Ok(exchange)
            })
        })
        // Load Balance: RoundRobin distributes messages across endpoints
        .load_balance()
        .to("log:server-a?showBody=true")
        .to("log:server-b?showBody=true")
        .to("log:server-c?showBody=true")
        .end_load_balance()
        .build()?;

- id: load-balancer-demo
  from: timer:tick?period=1000&repeatCount=10
  steps:
    - set_body:
        value: hello from load balancer
    - load_balance:
        strategy: round_robin
        steps:
          - to: log:server-a?showBody=true
          - to: log:server-b?showBody=true
          - to: log:server-c?showBody=true

The example sets a text body, then opens a load-balance block over three log endpoints with the default round_robin strategy. For each exchange, the strategy picks one endpoint and dispatches the exchange there. The other two endpoints see nothing. With ten timer ticks over three endpoints, round-robin hands ticks out in order. The first goes to log:server-a, the second to log:server-b, the third to log:server-c. The fourth cycles back to server-a.

The strategy decides which endpoint a given exchange hits. RoundRobin (the default) cycles the endpoints in order with a shared counter. Random picks an index at random on each call. Weighted takes a list of (name, weight) pairs and draws an endpoint in proportion to its weight. A heavier endpoint gets more traffic. Failover tries the endpoints in order. On error it moves to the next, and keeps going until one succeeds or the list is exhausted. Pick round-robin for uniform endpoints of equal capacity. Pick weighted when endpoints differ in throughput. Pick failover when one primary endpoint should serve and the rest stand by as backups.

The Load Balancer sends each exchange to one endpoint. Multicast sends a copy to all of them. A Content-Based Router picks a branch by predicate. The Load Balancer picks by strategy, not by message content. The selection is stateless across exchanges except for the round-robin counter. Under Random or Weighted, two consecutive exchanges may land on the same endpoint.

Per ADR-0025, the load balancer is an outcome-aware structural EIP. A selected branch that returns PipelineOutcome::Stopped propagates that stop out of the block. Under Failover, a branch that returns Failed moves the dispatch to the next endpoint instead of propagating the failure. Per ADR-0001, the block compiles into a Service<Exchange> step in the Tower pipeline, with each to arm a child step on the route channel. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/load-balancer.

Transformation

Transformation patterns change the content, format, or type of the exchange body. They convert between wire formats, enrich the body with external data, or set the body from an expression. They correspond to the Message Transformation category in Hohpe and Woolf.

• Convert Body — convert the exchange body to a new type in the pipeline
• Marshal and Unmarshal — serialize and deserialize the body to or from a named format
• Transform — set the body from a Simple expression or a literal value
• Script — run an inline script that can modify the exchange
• Poll Enrich — poll a resource and replace the body with the result
• Content Enricher — call a resource and replace the body with the enriched result

For the data types these steps operate on, see Exchange and Message.

Convert Body

The Convert Body step is a Message Translator (Hohpe & Woolf). It converts the exchange body to a target BodyType variant so downstream steps read the same data under a different type.

    let route_text_to_json = RouteBuilder::from("timer:text-to-json?period=2000&repeatCount=3")
        .route_id("text-to-json")
        .set_body(r#"{"message": "hello from text", "count": 42}"#)
        .log(
            "Route 1: Starting with Text body containing JSON string",
            LogLevel::Info,
        )
        .convert_body_to(BodyType::Json)
        .log(
            "Route 1: Converted Text → Json successfully!",
            LogLevel::Info,
        )
        .to("log:info?showBody=true")
        .error_handler(ErrorHandlerConfig::log_only())
        .build()?;

- id: text-to-json
  from: timer:text-to-json?period=2000&repeatCount=3
  error_handler:
    retry:
      max_attempts: 0
  steps:
    - set_body:
        value: '{"message": "hello from text", "count": 42}'
    - convert_body_to: json
    - to: log:info?showBody=true

The .convert_body_to(BodyType::Json) call picks a target variant from the BodyType enum: Json, Bytes, or Text. The step reads the current body, re-encodes it to that variant, and stores the result on the exchange. Steps after the conversion then read the body under its new type.

Use Convert Body when the data is correct but the type is wrong. A source that emits bytes and a sink that expects JSON sit at a type boundary. The step changes the type and keeps the content. Transform does a different job. It replaces the body from an expression or a literal value. Convert Body re-types the existing data. Transform overwrites it.

Convert Body works only within the built-in BodyType variants. Marshal and Unmarshal handle named wire formats such as CSV or Protobuf. Those steps translate between a structured type and a serialized representation, not between body types.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower middleware pipeline. The processor contract and BodyType definition are documented in camel-processor/CONTEXT.md.

The example source is at examples/convert-body-to.

Marshal and Unmarshal

Marshal and Unmarshal are the Message Translator pair (Hohpe & Woolf). Marshal serializes the body to a named wire format. Unmarshal deserializes a wire-format body back into a structured type. Together they cross the boundary between the pipeline and external systems.

    let route_csv_marshal = RouteBuilder::from("timer:csv-marshal?period=2000&repeatCount=3")
        .route_id("csv-marshal")
        .set_body(r#"[{"name":"Carol","age":28},{"name":"Dave","age":35}]"#)
        .unmarshal("json")?
        .log("Route 2: starting CSV marshal from Json", LogLevel::Info)
        .marshal("csv")?
        .log("Route 2: marshalled Json -> CSV Text", LogLevel::Info)
        .to("log:info?showBody=true")
        .error_handler(ErrorHandlerConfig::log_only())
        .build()?;

    let route_json_roundtrip = RouteBuilder::from("timer:json-roundtrip?period=2000&repeatCount=3")
        .route_id("json-roundtrip")
        .set_body(r#"{"message": "hello", "count": 42}"#)
        .log(
            "Route 1: Starting with Text body containing JSON string",
            LogLevel::Info,
        )
        .unmarshal("json")?
        .log("Route 1: Unmarshalled Text -> Json", LogLevel::Info)
        .marshal("json")?
        .log(
            "Route 1: Marshalled Json -> Text (round-trip complete!)",
            LogLevel::Info,
        )
        .to("log:info?showBody=true")
        .error_handler(ErrorHandlerConfig::log_only())
        .build()?;

- id: json-roundtrip
  from: timer:json-roundtrip?period=2000&repeatCount=3
  error_handler:
    retry:
      max_attempts: 0
  steps:
    - set_body:
        value: '{"message": "hello", "count": 42}'
    - unmarshal: json
    - marshal: json
    - to: log:info?showBody=true

- id: csv-marshal
  from: timer:csv-marshal?period=2000&repeatCount=3
  error_handler:
    retry:
      max_attempts: 0
  steps:
    - set_body:
        value: '[{"name":"Carol","age":28},{"name":"Dave","age":35}]'
    - unmarshal: json
    - marshal: csv
    - to: log:info?showBody=true

The .marshal("csv") call names the wire format as a string. The processor looks up that name in the data format registry and applies the format to the current body. Marshal stores the serialized result on the exchange. Unmarshal reverses the flow. It parses the body through the named format and stores the parsed structure on the exchange.

The string parameter keeps the route declaration format-agnostic. Common formats include json, csv, xml, and protobuf. Each format owns its body-type mapping and its configuration. See Data Formats for the format catalog, the DataFormat trait, and per-format options. This page covers only the route-level step.

A route that crosses a system boundary pairs the two steps. Marshal prepares the body for the wire on the outgoing side. Unmarshal restores a structured type on the incoming side. Each step in between then reads the body shape it expects.

Per ADR-0001, both steps compile into Service<Exchange> services in the Tower middleware pipeline. The data format registry and the marshal/unmarshal hooks are documented in camel-processor/CONTEXT.md.

The example sources are at examples/marshal-csv and examples/marshal-unmarshal.

Transform

The Transform step is a Message Translator from Hohpe and Woolf. It sets the exchange body from a literal value, a Simple expression, or a Rhai expression. In the Rust builder API, transform is an alias for set_body. Both compile to the same SetBody processor.

    let route = RouteBuilder::from("timer:transform?period=1000&repeatCount=3")
        .route_id("transform-pipeline")
        .process(|mut exchange| async move {
            exchange.input.body = Body::Text("hello world".to_string());
            Ok(exchange)
        })
        // .transform() is an alias for .set_body() — sets body to a static value
        .transform(Value::String("hello world (via transform)".into()))
        .map_body(|body: Body| {
            if let Some(text) = body.as_text() {
                Body::Text(text.to_uppercase())
            } else {
                body
            }
        })
        .set_header("transformed", Value::Bool(true))
        .to("log:info?showHeaders=true&showBody=true&showCorrelationId=true")
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

- id: transform-demo
  from: timer:tick?period=1000
  steps:
    - set_header:
        key: prefix
        value: hello
    - transform:
        simple: "${header.prefix} world"
    - to: log:transformed?showBody=true

The Rust .set_body(value) call accepts any type that implements Into<Body>. Strings, JSON values, and byte vectors all qualify. The step discards the old body and stores the new value on the exchange. When the new body depends on the current exchange, call .set_body_fn(closure) instead. The closure receives an &Exchange and returns a Body, so it can read headers, properties, and the inbound body to compute the replacement. The .transform(value) method forwards to set_body and exists for parity with the Apache Camel route DSL.

The YAML transform: step accepts three shapes that map to the SetBodyConfig fields in the DSL layer. A literal (transform: "hello") stores the value as-is. A Simple expression (transform: { simple: "${body.upper()}" }) evaluates the expression and stores the result. A Rhai expression (transform: { rhai: "body + '_processed'" }) does the same through the Rhai engine. The Rust API splits the static and dynamic cases across set_body and set_body_fn. The YAML form unifies them under one step.

Transform overwrites the body. Convert Body re-types the existing body without changing its content. A route that needs new content uses Transform. A route that needs the same content under a different type (Text to Json) uses Convert Body.

Transform evaluates one expression and writes one result. Script runs a full Rhai block that can mutate headers, properties, and body in the same step. A route that needs a single replacement uses Transform. A route that needs branching logic or several mutations uses Script.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/transform-pipeline.

Script

The Script step is a Message Translator from Hohpe and Woolf. It runs an inline Rhai script against the exchange so the route can mutate headers, properties, and body in one step. rust-camel registers Rhai as the scripting language.

        // Step 4: .script() — mutating Rhai expression tags the order and
        //         appends a status suffix. Changes propagate back to the Exchange.
        .script(
            "rhai",
            r#"
            headers["processed"] = true;
            let status = if headers["priority"] == "high" { "PRIORITY" } else { "STANDARD" };
            body = body + " [" + status + "]";
            "#,
        )
        // Step 5: log every message (after enrichment)
        .to("log:all-orders?showBody=true&showHeaders=true")

- id: script-demo
  from: timer:tick?period=1000
  steps:
    - script:
        language: rhai
        source: |
          headers["tenant"] = "acme"
          body = body + "_processed"
    - to: log:scripted?showBody=true&showHeaders=true

The .script("rhai", source) call hands the script text to the Rhai engine. The engine exposes three variables on the exchange: headers and properties as mutable maps, and body as the current body string. The script reads and writes all three. Assignments to headers["key"] and body propagate to the next step in the pipeline.

Script handles logic that one expression cannot express. A Rhai block runs conditionals, loops, and several mutations in a single step. Transform evaluates one expression and writes one result. A route that needs a single body replacement uses Transform. A route that needs branching logic or several field changes uses Script.

Script interprets Rhai at runtime, so it runs slower than a native .process(|ex| ...) closure. The closure compiles to Rust and reads the full Exchange type. A route that needs maximum throughput uses process. A route that needs logic it can change without a rebuild uses Script.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower middleware pipeline. The Rhai integration is documented in camel-language-api/CONTEXT.md.

The example source is at examples/language-rhai.

Poll Enrich

The Poll Enrich is a Content Enricher variant from Hohpe and Woolf. It polls a passive resource and feeds the result into the exchange.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=3")
        .route_id("pollenrich-demo")
        .poll_enrich(
            format!("file:{config_dir_str}?fileName=config.json&noop=true"),
            5000,
        )
        .stream_cache_default()
        .to("log:enriched?showBody=true&showHeaders=true&showCorrelationId=true")
        .build()?;

- id: pollenrich-demo
  from: timer:tick?period=1000&repeatCount=3
  steps:
    - poll_enrich:
        uri: "file:/tmp/rust-camel-pollenrich?fileName=config.json&noop=true"
        timeout: 5000
    - stream_cache: true
    - to: log:enriched?showBody=true&showHeaders=true&showCorrelationId=true

The .poll_enrich(uri, timeout) step reads from a polling consumer. It waits up to timeout milliseconds for a message. A file or database is a passive resource. It stores data but does not push it. Poll Enrich pulls that data on demand. The example reads a config file each time the timer fires.

The EnrichmentStrategy trait decides how the polled exchange combines with the original. The default UseEnrichedBody strategy replaces the original body with the polled body. It keeps the original headers and properties. It discards the headers from the polled exchange. Write a custom strategy when you need to merge both bodies or preserve the polled headers.

When the poll returns no message within the timeout, the step calls on_no_poll. The default behavior passes the original exchange through unchanged. ThrowOnNoPoll wraps a base strategy and errors the exchange instead. Use it when missing data is a failure, not a normal condition.

Choose Poll Enrich when the source is a passive consumer. Choose the Content Enricher when the source is an active endpoint that receives a request and returns a response. Poll Enrich reads. Content Enricher calls.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower pipeline. The strategy contract and the PollEnrichService implementation are documented in camel-processor/CONTEXT.md.

Reference: Processor crate · Example source

Content Enricher

The Content Enricher is a Message Translator from Hohpe and Woolf. It calls a producer endpoint and feeds the response into the exchange.

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=3")
        .route_id("content-enricher-demo")
        .enrich("direct:enrich-data")
        .to("log:enriched?showBody=true&showCorrelationId=true")
        .build()?;

- id: content-enricher-demo
  from: timer:tick?period=1000&repeatCount=3
  steps:
    - enrich: "direct:enrich-data"
    - to: log:enriched?showBody=true&showCorrelationId=true

- id: enrichment-source
  from: direct:enrich-data
  steps:
    - set_body:
        value: "enriched-value"

The .enrich(uri) step sends the exchange to a producer endpoint and waits for a response. The example routes the call to direct:enrich-data. A second route consumes from that endpoint and supplies the enrichment payload. The two-route pattern keeps the enrichment source out of the main route. Multiple consumers can share the same source.

The EnrichmentStrategy trait decides how the response combines with the original exchange. The default UseEnrichedBody strategy replaces the original body with the response body. It keeps the original headers and properties. It discards the response headers. Write a custom strategy when you need to merge both bodies or preserve the response headers.

The difference from Poll Enrich is the data source. Content Enricher calls an active endpoint that receives a request and returns a response. Poll Enrich reads from a polling consumer that holds passive data. Use Content Enricher to pull data from a service. Use Poll Enrich to read from a file or database.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower pipeline. The EnrichService calls the producer inside the step and merges the response before the step returns. The processor contract is documented in camel-processor/CONTEXT.md.

Reference: Processor crate · Example source

Messaging

Messaging patterns change the cardinality or order of exchanges. They split one message into many, gather many into one, reorder a sequence, or sample one in every N. Several compose into split-process-aggregate flows.

• Aggregator — collect exchanges by correlation key and emit a batch
• Splitter — break a composite message into one exchange per fragment
• Streaming Splitter — split a byte-stream body into fragment exchanges with backpressure
• Zip Splitter — split a ZIP archive into one exchange per entry
• Resequencer — reorder exchanges by sequence number in batches or streams
• Sort — sort an array body by a comparator expression
• Sampling — pass through one out of every N exchanges
• Claim Check — stash a large payload and replace it with a claim ticket
• Cache — cache a computed body by key with TTL and stale-read fallback

For the processor contract that implements these patterns, see camel-processor/CONTEXT.md.

Aggregator

The Aggregator is a Message Routing pattern from Hohpe and Woolf. It collects related exchanges into a bucket and emits one combined exchange when a completion condition holds.

    let route = RouteBuilder::from("timer:orders?period=200&repeatCount=12")
        .route_id("aggregator-demo")
        .process(move |mut ex: camel_api::Exchange| {
            let c = Arc::clone(&counter_clone);
            Box::pin(async move {
                let n = c.fetch_add(1, Ordering::SeqCst);
                let order_id = ["A", "B", "C"][(n % 3) as usize];
                ex.input
                    .headers
                    .insert("orderId".to_string(), serde_json::json!(order_id));
                ex.input.body = Body::Text(format!("order-item-{n}"));
                println!("[timer] #{n} orderId={order_id}");
                Ok(ex)
            })
        })
        .aggregate(
            AggregatorConfig::correlate_by("orderId")
                .complete_when_size(3)
                .max_buckets(100)
                .bucket_ttl(std::time::Duration::from_secs(60))
                .build()
                .unwrap(), // allow-unwrap
        )
        // Pending exchanges (Body::Empty, CamelAggregatorPending=true) still flow
        // through the pipeline — log only completed batches.
        .process(|ex: camel_api::Exchange| {
            Box::pin(async move {
                if ex.property("CamelAggregatorPending").is_some() {
                    return Ok(ex);
                }
                let key = ex
                    .property("CamelAggregatedKey")
                    .map(|v: &serde_json::Value| v.to_string())
                    .unwrap_or_default();
                let size = ex
                    .property("CamelAggregatedSize")
                    .map(|v: &serde_json::Value| v.to_string())
                    .unwrap_or_default();
                println!(
                    "[batch] orderId={} size={} body={:?}",
                    key, size, ex.input.body
                );
                Ok(ex)
            })
        })
        .to("log:batch?showBody=true&showCorrelationId=true")
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

- id: aggregator-demo
  from: timer:orders?period=200&repeatCount=12
  error_handler:
    retry:
      max_attempts: 1
  steps:
    - aggregate:
        header: "orderId"
        completion_size: 3
        max_buckets: 100
        bucket_ttl_ms: 60000
    - to: log:batch?showBody=true&showCorrelationId=true

Each incoming exchange carries a correlation key in a header. The correlate_by("orderId") call names that header. Exchanges that share the key land in the same bucket. An AggregationFn folds each new exchange into the bucket seed to build the emitted batch body. In the included route, the process step rotates the orderId header through "A", "B", and "C", so three buckets fill in parallel.

A bucket completes when it reaches its size limit or when its inactivity timeout fires. complete_when_size(3) flushes the bucket after three exchanges arrive for that key. complete_when_timeout(Duration) flushes it after a period with no new exchange. The bucket_ttl setting caps how long an incomplete bucket can live before the background sweep evicts it. The config validator rejects any setup with no memory bound, so set max_buckets, a timeout, or bucket_ttl.

Exchanges that arrive before a bucket completes still pass through the pipeline. They carry the CamelAggregatorPending property and an empty body. The process step after the aggregator checks that property and skips them. Only completed batches carry the CamelAggregatedKey and CamelAggregatedSize properties.

Use the Aggregator when many correlated messages arrive over time and you want one combined output. Multicast does the opposite job. It sends one message to many endpoints at once. The Aggregator collects. Multicast fans out.

Per ADR-0001, the aggregator compiles into a Service<Exchange> step in the Tower middleware pipeline. The processor contract and the divergences from Apache Camel are documented in camel-processor/CONTEXT.md.

The example source is at examples/aggregator.

Splitter

The Splitter is a Message Routing pattern from Hohpe and Woolf. It takes one composite message and produces one exchange per fragment.

    let route = RouteBuilder::from("timer:batch?period=2000&repeatCount=3")
        .route_id("splitter-demo")
        // Simulate incoming CSV data
        .process(|mut exchange: camel_api::Exchange| {
            Box::pin(async move {
                exchange.input.body = Body::Text("alice,100\nbob,200\ncharlie,300".to_string());
                Ok(exchange)
            })
        })
        // Split by lines, aggregate all fragment bodies into a JSON array
        .split(SplitterConfig::new(split_body_lines()).aggregation(AggregationStrategy::CollectAll))
        // Transform each CSV line into a JSON object
        .map_body(|body: Body| {
            let text = body.as_text().unwrap_or("");
            let parts: Vec<&str> = text.splitn(2, ',').collect();
            let (name, amount) = match parts.as_slice() {
                [n, a] => (*n, a.parse::<u64>().unwrap_or(0)),
                _ => (text, 0),
            };
            Body::Json(serde_json::json!({
                "name": name,
                "amount": amount,
            }))
        })
        .to("log:fragment?showBody=true&showCorrelationId=true")
        .end_split()
        // After split: aggregated JSON array of all fragments
        .to("log:aggregated?showBody=true&showCorrelationId=true")
        .error_handler(
            ErrorHandlerConfig::log_only()
                .on_exception(|_| true)
                .retry(1)
                .build(),
        )
        .build()?;

- id: splitter-demo
  from: timer:batch?period=2000&repeatCount=3
  error_handler:
    retry:
      max_attempts: 1
  steps:
    - set_body:
        value: "alice,100\nbob,200\ncharlie,300"
    - split:
        expression: body_lines
        aggregation: collect_all
        steps:
          - to: log:fragment?showBody=true&showCorrelationId=true
    - to: log:aggregated?showBody=true&showCorrelationId=true

A split expression decides how to divide the body. The included route uses split_body_lines(), which returns one fragment per line. Each fragment becomes its own exchange that flows through the sub-pipeline. A fragment inherits the parent headers, properties, message pattern, and OpenTelemetry context, so its span is a child of the parent span. The CamelSplitIndex, CamelSplitSize, and CamelSplitComplete properties mark where each fragment sits in the batch. The .map_body(...) step inside the split scope turns each line into a JSON object, and .end_split() closes the scope.

When the split scope closes, an aggregation strategy combines the fragment outputs. CollectAll gathers every fragment body into a JSON array. That array then flows to the step after .end_split(). Per the aggregation contract, failed fragments reach the strategy as Err(e) entries in the vector, not as exchanges with an attached exception.

The Splitter differs from Multicast in how it produces children. Multicast sends the same exchange to several endpoints. The Splitter derives one exchange per fragment from a single input. It also pairs with the Aggregator in a split-process-aggregate flow. The Splitter divides one message into many. The Aggregator collects many correlated messages back into one.

Per ADR-0001, the splitter compiles into a Service<Exchange> step in the Tower middleware pipeline. The per-fragment sub-pipeline compiles into child steps on the same route channel. The processor contract is documented in camel-processor/CONTEXT.md.

Per ADR-0025, the split block is an outcome-aware structural EIP. A Stopped outcome from the sub-pipeline returns Stopped(fragment_ex) and skips aggregation. The parent pipeline never sees a partial batch. The per-fragment boundary keeps the fragment mutations made before the Stop visible to the outer pipeline.

The example source is at examples/splitter.

Streaming Splitter

The Streaming Splitter is a Message Routing pattern from Hohpe and Woolf. It splits a streaming body into individual exchanges one fragment at a time, without first buffering the full body in memory.

    let mut splitter = StreamingSplitterService::new(
        expression,
        sub_pipeline,
        AggregationStrategy::CollectAll,
        true,
    );

    let ndjson = vec![
        r#"{"user":"alice","action":"login"}"#,
        r#"{"user":"bob","action":"purchase"}"#,
        r#"{"user":"charlie","action":"logout"}"#,
    ];
    let exchange = Exchange::new(Message::new(make_ndjson_stream(ndjson)));

    println!("Streaming Split example");
    println!("Input: Body::Stream with 3 NDJSON lines");
    println!();

    let _ = splitter.poll_ready(&mut std::task::Context::from_waker(
        &futures::task::noop_waker(),
    ));
    let result = splitter.call(exchange).await?;

    println!();
    println!("Aggregated result body:");
    println!("  {:?}", result.input.body);

- id: streaming-split-demo
  from: file:data.ndjson
  steps:
    - split:
        streaming: true
        stream:
          format: ndjson
        aggregation: collect_all
        steps:
          - to: log:fragment?showBody=true&showCorrelationId=true
    - to: log:aggregated?showBody=true&showCorrelationId=true

The included example builds a Body::Stream that holds three NDJSON chunks. A StreamingSplitterService reads the stream through a StreamSplitCodec, which resolves the format from the content type. For application/x-ndjson, the codec parses each line into a separate fragment exchange. The sub-pipeline logs each fragment. When the split scope closes, the aggregation strategy combines the fragment outputs into the result body.

The streaming variant is the memory-efficient alternative to the Splitter. The Splitter materializes every fragment before it processes the first one. The Streaming Splitter pulls one fragment from the source, runs the sub-pipeline, then pulls the next. A multi-gigabyte NDJSON file or a long-running log stream fits in constant memory. The codec reads bytes lazily, so the source produces data only as fast as the sub-pipeline accepts it.

Backpressure flows through the segment boundary. When the sub-pipeline pauses, the segment stops pulling from the stream, and the source stops producing. A Stopped outcome drops the underlying stream and returns the fragment exchange to the outer pipeline. The outer pipeline sees the same outcome shape it would from the eager Splitter.

Per ADR-0025, streaming split is an outcome-aware structural EIP. Stop and Failed outcomes flow through the CompiledStep::Segment boundary with the fragment exchange intact. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/streaming-split.

ZIP Splitter

The ZIP Splitter is a specialized Splitter from Hohpe and Woolf. It decomposes a multi-entry ZIP archive into one exchange per entry. Each entry flows through the route as its own message.

    let route = camel_builder::RouteBuilder::from("timer:zip-marshal?period=3000&repeatCount=2")
        .route_id("zip-marshal-route")
        .set_body("payload for ZIP compression")
        .log("Route 3: Original body", LogLevel::Info)
        .marshal("zip")?
        .log("Route 3: After marshal to ZIP", LogLevel::Info)
        .unmarshal("zip")?
        .log(
            "Route 3: After unmarshal from ZIP (round-trip!)",
            LogLevel::Info,
        )
        .to("log:zip-result?showBody=true")
        .error_handler(ErrorHandlerConfig::log_only())
        .build()?;

- id: zip-marshal-route
  from: timer:zip-marshal?period=3000&repeatCount=2
  error_handler: {}
  steps:
    - set_body: "payload for ZIP compression"
    - log: "Route 3: Original body"
    - marshal: zip
    - log: "Route 3: After marshal to ZIP"
    - unmarshal: zip
    - log: "Route 3: After unmarshal from ZIP (round-trip!)"
    - to: log:zip-result?showBody=true

The included route runs a timer that fires twice. The .marshal("zip") step compresses the body into a single-entry archive. The .unmarshal("zip") step decompresses it back. The route logs the byte sizes at each step. A reader can watch the body shrink during compression and grow during decompression.

The actual split happens through the zip_splitter expression and the StreamingSplitterService. The expression reads the ZIP central directory and emits one exchange per entry. Each exchange carries the entry body plus headers: CAMEL_ZIP_ENTRY_NAME, CAMEL_ZIP_ENTRY_PATH, CAMEL_ZIP_ENTRY_SIZE, and CAMEL_ZIP_ENTRY_INDEX. The ZipSplitConfig struct caps the entry count, the per-entry size, and the total decompressed size.

The ZIP Splitter is format-specific. It knows how to parse the ZIP archive structure. The Streaming Splitter is the generic alternative. It splits any byte stream through a codec that the content type selects. Use the ZIP Splitter when the input is a ZIP archive and you need per-entry headers. Use the Streaming Splitter for NDJSON, log lines, or raw byte chunks.

Per ADR-0001, the splitter compiles into a Service<Exchange> step in the Tower middleware pipeline. The per-entry sub-pipeline compiles into child steps on the same route channel. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/zip-splitter.

Sort

The Sort is a Message Translator from Hohpe and Woolf. It orders the elements of a body array by a sort key expression. The route downstream sees the array in the chosen order.

      - set_body:
          value: [3, 1, 4, 1, 5, 9, 2, 6]
      - sort:
          expression: "${body}"
      - to: "log:sorted?showBody=true&showCorrelationId=true"

The included route fires a timer every second for three ticks. The set_body step sets the body to a JSON array of eight numbers. The sort step orders the array by the ${body} expression, which uses each element itself as the sort key. The to step logs the sorted array. The sort defaults to ascending order. A reverse: true flag produces descending order.

The Sort step reads the body as a JSON array. A non-array body fails with a processor error. The upstream set_body step guarantees that contract for this example. The sort key expression extracts a comparable value from each element. The expression ${body} uses the element itself. The expression ${body.field} extracts a nested field for sorting objects by a property.

The Sort differs from the Splitter. The Sort keeps the body as a single array and reorders it in place. The Splitter decomposes the body into many exchanges. A route that needs both patterns sorts first and then splits per element. The Sort also pairs with the Aggregator. A split-sort-aggregate flow splits a collection, sorts the fragments, then aggregates the results back into one message.

Per ADR-0001, the sort step compiles into a Service<Exchange> step in the Tower middleware pipeline. The sort runs inside the step and the ordered body flows out before the next step runs. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/sort.

Sampling

The Sampling is a Message Router from Hohpe and Woolf. It passes one exchange out of every N and drops the rest. The route downstream sees a subsample of the upstream traffic.

      - sampling: 3
      - to: "log:passed?showBody=true&showCorrelationId=true"

The sample period is an integer. sampling: 3 accepts one exchange out of every three and drops the other two. The counter starts at zero. The step increments it before the modulo check. Exchange number three is the first one that passes, then six, then nine. A period of one passes every exchange. The step rejects a period of zero at build time.

The step sets the CamelStop header on a dropped exchange. The surrounding pipeline translates that header into PipelineOutcome::Stopped. The steps after the sampling step do not run for that exchange. The included route fires a timer nine times. The log step runs three times.

Use the Sampling when the downstream rate is more than the route needs. Load shedding drops traffic before it reaches expensive steps. A monitoring or debugging route can sample a feed so the logs stay readable under high volume. A period-N sampler reduces the rate by an integer factor without the state of a rate limiter.

The Sampling differs from the Throttler. The Sampling is stateless and uses a counter. The Throttler paces the rate over a sliding time window and queues excess exchanges. A route that needs both patterns samples first, then throttles.

Per ADR-0001, the sampling step compiles into a Service<Exchange> step in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/sampling.

Resequencer

The Resequencer is a Message Router from Hohpe and Woolf. It reorders a stream of exchanges into a defined sequence. The route downstream sees the exchanges in order even when the upstream side delivers them out of order.

      - set_header:
          key: "seq"
          simple: "${header.CamelTimerCounter}"
      - set_header:
          key: "region"
          value: "default"
      - resequence:
          batch:
            correlation: "${header.region}"
            sort: "${header.seq}"
            completion:
              size_or_timeout: [3, 2000]
      - to: "log:sorted?showBody=true&showCorrelationId=true"

The batch policy buffers exchanges per correlation key. Each exchange enters the bucket that matches its correlation header. The bucket completes when it reaches its size limit or when the timeout window elapses. At completion, the policy sorts the buffered exchanges by the sort expression. It emits the sorted exchanges as an ordered burst. The included route copies the timer counter into a seq header and tags each exchange with a region correlation key. The resequence.batch step releases three sorted exchanges at a time.

The sort expression extracts a comparable value from each exchange. The route uses ${header.seq}, so the release order matches the sequence numbers. Exchanges that arrive out of order wait in the buffer until the window completes. The policy then sorts them into the correct sequence before release.

The Resequencer differs from the Sort. The Sort orders the elements of one body array in place. The Resequencer buffers separate out-of-order exchanges across many messages, then releases them in sequence. Use Sort when one exchange carries a collection. Use Resequencer when many exchanges arrive out of sequence. The Resequencer also differs from the Aggregator. The Aggregator fuses many exchanges into one. The Resequencer keeps one exchange per output and only changes the order. A route that needs both patterns places a Resequencer before an Aggregator.

Per ADR-0029, the resequencer is a continuation boundary. The main pipeline ends at the resequencer. A post-driver task runs the steps after the resequencer on each sorted emission. Per ADR-0025, the batch and stream policies propagate PipelineOutcome::Stopped through the post-continuation. Per ADR-0001, the pre-steps compile into the main Service<Exchange> pipeline and the post-steps compile into a continuation processor. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/resequencer.

Claim Check

The Claim Check is a Message Translator from Hohpe and Woolf. It stores the full message payload in a repository and replaces the body with a lightweight claim key. The route carries only the key until a later step retrieves the original body.

      - claim_check:
          repository: vault
          operation: set
          key: "${header.claimKey}"
      - log: "After SET body is the claim key: ${body}"
      - claim_check:
          repository: vault
          operation: get
          key: "${header.claimKey}"
      - log: "After GET body restored: ${body}"

The claim_check step delegates to a ClaimCheckRepository registered on the context. The set operation stashes the current body under the given key and replaces the body with that key. The get operation reads the key, retrieves the stashed body from the repository, and restores it as the body. The included route sets a placeholder payload, pins the claimKey header, then calls set to stash the body. The log step after set shows the claim key as the new body. The second claim_check step calls get with the same key, and the final log step shows the restored body.

Use the Claim Check when the payload is large and the route does not need the full body at every step. Store-and-forward flows stash the payload at the ingress. The route carries only the lightweight key through the pipeline. A later step retrieves the payload at the egress before delivery. This keeps the in-memory footprint low across long routes. The repository also supports get_and_remove, push, and pop for stack and queue access patterns.

The Claim Check differs from the Idempotent Consumer. Both patterns use a repository trait for their state. The Idempotent Consumer stores only the deduplication key. The Claim Check stores the full payload. A route that needs to deduplicate and stash uses both patterns. Claim Check alone keeps a heavy payload out of the pipeline.

Per ADR-0001, the claim check step compiles into a Service<Exchange> step in the Tower middleware pipeline. The repository trait is defined in ADR-0028. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/claim-check.

Cache

The Cache pattern stores the result of an expensive computation and serves it on subsequent requests with the same key. When the key matches, the route returns the cached body and skips the computation. When the key does not match, the route runs the on_miss sub-pipeline, stores the result, and returns it.

      - cache:
          key: "${header.cacheKey}"
          ttl: "5s"
          on_miss:
            - set_body: "computed-fresh-data"
            - log: "Cache MISS — body computed and stored"
      - log: "After cache body is: ${body}"

The cache step evaluates a key expression against each exchange. If the repository holds a live entry for that key, the step replaces the body with the cached bytes and returns Completed. If the key is absent or expired, the step runs the on_miss sub-pipeline, writes the resulting body to the repository with the given ttl, and returns the body to the pipeline. A key expression that evaluates to None bypasses the cache. The route runs the on_miss sub-pipeline without a lookup or a write-back.

The cache_invalidate step removes a single key from the repository. Use it when an upstream event makes a cached entry stale. The cache_peek_stale step reads a cached entry and ignores its in-band expiry. This serves a post-expiry entry as a fallback when the source is unavailable.

Use the Cache pattern when a route computes the same result more than once. API responses, database lookups, and transform-heavy pipelines benefit from caching. Pair cache_peek_stale with a Circuit Breaker to serve stale data when the downstream service is open.

The default repository is "memory" (moka-backed, size-eviction only). A persistent "persistent" repository (redb-backed) is available when [default.cache_repo] backend = "redb" is set. The redb backend survives process restarts. Its sweep task reclaims entries whose expires_at + stale_retention has passed. The memory backend does not run a sweep. Expired entries stay in memory until size pressure evicts them.

The Cache differs from the Claim Check and the Idempotent Consumer. All three use a repository trait. The Cache stores the full computed body with a TTL. The Claim Check stores the original payload without a TTL. The Idempotent Consumer stores only the deduplication key. A route that needs all three can chain them.

Per ADR-0056, the CacheRepository trait lives in camel-api, with memory and redb backends in camel-core. The trait stores CacheEntry { bytes, content_type, expires_at } with in-band expiry. Both backends do size-eviction only. The expires_at field drives get() misses and peek_stale() reads. Per ADR-0001, each cache step compiles into a Service<Exchange> step in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/cache-example.

Resilience and control

Resilience and control patterns protect a route from failure, limit throughput, and scope error handling. They map to the System Management family in Hohpe and Woolf, adapted to the Tower middleware pipeline.

• Circuit Breaker — trip after repeated failures and recover after a cool-down
• Do Try — scoped catch and finally blocks around a group of steps
• Throttler — cap the rate of exchanges through a route
• Idempotent Consumer — reject duplicates by correlation key
• Delayer — hold an exchange for a fixed duration
• Loop — repeat a sub-route a fixed number of times
• Validator — check exchange content against a schema

For the error-handling contract that governs catch and finally behavior, see error handling.

Circuit breaker

The Circuit Breaker is a System Management pattern from Hohpe and Woolf. It trips after repeated failures against a downstream service, then short-circuits further calls for a cool-down so the dependency can recover.

    let main_route = RouteBuilder::from("timer:cb-test?period=1000&repeatCount=15")
        .route_id("circuit-breaker-demo")
        .process(|mut exchange| async move {
            let n = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            exchange.input.body = Body::Text(format!("request-{}", n));
            Ok(exchange)
        })
        .to("direct:failing-service")
        .circuit_breaker(
            CircuitBreakerConfig::new()
                .failure_threshold(3)
                .open_duration(Duration::from_secs(3)),
        )
        .to("log:cb-success?showBody=true&showCorrelationId=true")
        .error_handler(
            ErrorHandlerConfig::dead_letter_channel(
                "log:cb-fallback?showBody=true&showHeaders=true&showCorrelationId=true",
            )
            .on_exception(|_| true)
            .build(),
        )
        .build()?;

- id: circuit-breaker-demo
  from: timer:cb-test?period=1000&repeatCount=15
  circuit_breaker:
    failure_threshold: 3
    open_duration_ms: 3000
  error_handler:
    dead_letter_channel: log:cb-fallback?showBody=true&showHeaders=true&showCorrelationId=true
  steps:
    - to: direct:failing-service
    - to: log:cb-success?showBody=true&showCorrelationId=true

The breaker cycles through three states. In Closed, traffic flows. Each failed call increments a consecutive-failure counter. A successful call resets that counter to zero. When the counter reaches failure_threshold, the breaker trips into Open. In Open, the breaker rejects every call with CamelError::CircuitOpen. The route never touches the downstream service. The breaker holds Open for open_duration, then enters HalfOpen.

HalfOpen admits a single probe call. The breaker rejects concurrent callers in that window so the probe runs in isolation. A probe that succeeds closes the breaker and resets the counter. A probe that fails reopens the breaker for another full cool-down. This single-probe design stops a backlog of traffic from stampeding the service. The dependency gets one probe, not a flood, at the first sign of recovery.

Use the circuit breaker to protect a downstream service. Do not use it to repair a transient blip. Retry sends more traffic at a failing call. The circuit breaker sends none. Pair them when a flaky dependency needs a retry for transient errors but a hard stop during a sustained outage. Per ADR-0019, the breaker compiles into a gate on RouteChannelService rather than a pipeline step. Boundary rejections flow through RouteErrorHandler::handle_boundary. The route's error_handler can then route CircuitOpen to a dead-letter sink, as the example does with log:cb-fallback.

The example source is at examples/circuit-breaker.

Do Try

The Do Try is an Error Handling pattern from Hohpe and Woolf. It wraps a group of steps in a local scope with one or more catch clauses and an optional finally clause. A route can repair or clean up after a failing step without triggering the route-level error handler.

    let route1 = RouteBuilder::from("direct:catch-by-variant")
        .route_id("catch-by-variant")
        .do_try()
        .process(always_fail("boom-from-route-1"))
        .do_catch_exception(&["ProcessorError"])
        .handled()
        .process(log_marker("log:caught-by-variant"))
        .end_do_catch()
        .end_do_try()
        .build()?;

- id: catch-by-variant
  from: direct:catch-by-variant
  steps:
    - do_try:
        steps:
          - to: direct:failing-op
        catch:
          - exception:
              - ProcessorError
            steps:
              - to: log:caught-by-variant

A do_try() block opens the scope. Steps inside the try body run in sequence. When a step returns Err, the block walks its catch clauses in order and runs the body of the first match. do_catch_exception(&["ProcessorError"]) matches by CamelError variant name. do_catch_when(predicate) matches by a FilterPredicate over the exchange when the variant alone is not specific enough. The example shows both shapes. List specific clauses before broad ones. The first match wins.

do_finally() adds a body that runs exactly once whether the try body succeeded, threw, or was caught. Use it for cleanup that must always run. The example's third route pairs a .propagate() catch with a do_finally counter. The catch logs the failure and lets the error escape. The finally block still runs and bumps the counter.

Each catch clause ends with a disposition. The disposition decides what happens to the exchange after the catch body runs. Handled clears the error and stops the block. propagate() keeps the error live so it escapes after the catch body finishes. The full disposition model, with all values and their effects on the pipeline, lives in error handling. The block is a local error-handling island. A Handled catch never reaches the route's error_handler. Only unhandled errors and errors from a propagate() catch escape to the route level.

Use do_try when the repair is scoped to one step or a small group of steps. Use the route-level error_handler when every step in the route shares one retry, dead-letter, or disposition policy. The two compose. A Handled catch repairs a step locally. Unhandled failures still fall through to the route-level safety net.

Per ADR-0001, the do-try compiles into a Service<Exchange> step in the Tower middleware pipeline. The try body and each catch branch compile as child steps on the same route channel. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/do-try.

Throttler

The Throttler is a System Management pattern from Hohpe and Woolf. It caps how many exchanges a route processes in a fixed time window. Downstream services never receive traffic faster than they can handle.

    let route = RouteBuilder::from("timer:tick?period=100&repeatCount=20")
        .route_id("throttler-demo")
        // Throttle: limit to 2 requests per second
        // Default strategy is Delay - queues messages until capacity available
        .throttle(2, Duration::from_secs(1))
        .to("log:throttled?showBody=true")
        .end_throttle()
        .build()?;

- id: throttler-demo
  from: timer:tick?period=100&repeatCount=20
  steps:
    - throttle:
        max_requests: 2
        period_secs: 1
        strategy: delay
        steps:
          - to: log:throttled?showBody=true

The .throttle(2, Duration::from_secs(1)) call sets the rate limit. At most two exchanges pass per second. The included route fires a timer 20 times at 100-millisecond intervals, which yields ten exchanges per second. The throttler holds the excess in an internal queue and releases those exchanges as the one-second window refills. Each released exchange flows into the .to("log:throttled?showBody=true") step inside the throttle scope. The .end_throttle() call closes the scope.

The default Delay strategy queues excess exchanges instead of dropping them. The throttler never discards an exchange on its own. This protects a downstream service from bursts without losing messages. A route that must reject instead of queue composes a filter on a backpressure signal.

Use the Throttler when a downstream service or external API imposes a rate limit. Database writers, third-party HTTP endpoints, and metered SaaS APIs reject or fail when traffic exceeds their quota. The Throttler smooths the source rate to fit that contract.

Per ADR-0001, the throttler compiles into a Service<Exchange> step in the Tower middleware pipeline. The rate-limit window lives inside the service. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/throttler.

Idempotent Consumer

The Idempotent Consumer is a System Management pattern from Hohpe and Woolf. It detects duplicate exchanges by a message key and skips them. The route processes only the first delivery of each key.

      - idempotent_consumer:
          repository: dedupe
          expression: "${header.messageId}"
          steps:
            - log: "Processing unique message ${header.messageId}"
            - to: "log:processed?showBody=true&showHeaders=true"

The idempotent_consumer step computes a key from each exchange with a MessageIdExpression. In the included route, the expression reads ${header.messageId}. The step then asks its repository whether it has seen the key before. A new key runs the child steps, and the step records that key in the repository. A repeated key skips the child steps entirely. The route pins messageId to a fixed value across five timer ticks. Only the first tick runs the log and to steps. The other four are duplicates.

A duplicate does not raise an error. The segment returns Completed and the parent pipeline continues. This matters when a source retries delivery on failure. Without deduplication, each retry re-runs the child steps and produces duplicate side effects. With the Idempotent Consumer, the second and later deliveries of the same key return Completed and leave the recorded result intact.

Use the Idempotent Consumer when a route must tolerate redelivery without repeating work. Payment processing, order creation, and any at-least-once message source benefit from a deduplication gate. The example registers a memory-backed repository. A memory-backed repository loses its keys when the process exits, so choose a durable backing store when a route must survive a restart.

The Idempotent Consumer differs from the Claim Check. Both use a repository trait. The Idempotent Consumer stores only the key. The Claim Check stores the full payload. Per ADR-0025, the consumer is an outcome-aware segment. Per ADR-0001, it compiles into a Service<Exchange> step in the Tower middleware pipeline. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/idempotent-consumer.

Delayer

The Delayer is a System Management pattern from Hohpe and Woolf. It holds an exchange in the pipeline for a fixed or dynamic duration. The exchange then moves to the next step.

    let fixed_route = RouteBuilder::from("timer:tick?period=2000&repeatCount=5")
        .route_id("delayer-fixed")
        .delay(Duration::from_millis(500))
        .to("log:delayed?showBody=true")
        .build()?;

- id: delayer-fixed
  from: timer:tick?period=2000&repeatCount=5
  steps:
    - delay:
        delay_ms: 500
    - to: log:delayed?showBody=true

The included route fires a timer every two seconds. The .delay(Duration::from_millis(500)) step suspends the exchange for a fixed half-second. The log:delayed step receives the exchange only after the timer elapses. In the YAML DSL the same fixed pause is the delay_ms field. Use a fixed delay to space messages apart by a known interval. Common cases are rate-limited APIs and scheduled batches.

For a per-message pause, set the dynamic_header option to a header name. Write the delay in milliseconds to that header on each exchange. The service reads the header value and clamps it to max_delay_ms (default 3,600,000 ms, one hour). A missing or non-numeric header falls back to delay_ms. The builder equivalent is delay_with_header. The example registers it on a second route that sets CamelDelayMs to 1000. A dynamic delay suits routes where the producer supplies a retry-after or backoff value.

The Delayer differs from the Throttler. The Delayer pauses every exchange by the configured amount. The Throttler paces the rate of accepted exchanges over a window. A route that needs both can place a Delayer after a Throttler.

Per ADR-0001, the delayer compiles into a Service<Exchange> step. Its call method awaits a tokio::time::sleep and returns the exchange when the timer elapses. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/delayer.

Loop

The Loop is a System Management pattern from Hohpe and Woolf. It repeats a block of steps a fixed number of times for each incoming exchange. The body or state can change one iteration at a time.

    let loop_route = RouteBuilder::from("timer:tick?period=3000&repeatCount=3")
        .route_id("loop-count-demo")
        .set_body("hello")
        .loop_count(3)
        .process(|mut ex: camel_api::Exchange| async move {
            let body = ex.input.body.as_text().unwrap_or("").to_string();
            ex.input.body = camel_api::body::Body::Text(format!("{body}!"));
            Ok(ex)
        })
        .end_loop()
        .to("log:loop-result?level=info&showBody=true")
        .build()?;

- id: loop-count-demo
  from: timer:tick?period=3000&repeatCount=3
  steps:
    - set_body:
        value: hello
    - loop:
        count: 3
        steps:
          - to: log:loop-iteration
    - to: log:loop-result?level=info&showBody=true

The included route fires a timer three times. The .set_body("hello") step sets the body to a short string. The .loop_count(3) step opens the loop and .end_loop() closes it. The .process(...) step inside the loop appends one ! to the body on each pass. After three iterations the body becomes hello!!!. The log:loop-result step after the loop logs the final body. In the YAML DSL the loop count is the count field under loop. The count is fixed before the route starts.

Each iteration runs the wrapped sub-pipeline against the same exchange. A mutating step builds on the result of the previous pass. The loop acts as a fold over a fixed range. Use a loop for retry-with-backoff sequences, batch enrichment, or pagination that fetches one page per iteration. The Loop differs from the Splitter. The Loop produces a single output exchange with accumulated state. The Splitter produces many output exchanges, one per fragment.

Per ADR-0025, the loop is an outcome-aware structural EIP. A Stopped outcome from the inner sub-pipeline returns Stopped(ex) and skips the remaining iterations. The parent pipeline never sees a half-applied loop body. Per ADR-0001, the loop compiles into a Service<Exchange> step. The per-iteration sub-pipeline compiles into child steps on the same route channel. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/loop.

Validator

The Validator is a Message Transformation pattern from Hohpe and Woolf. It checks an exchange body against a schema or predicate. It rejects the exchange when the check fails, so the rest of the route only sees valid input.

    let route_xsd = RouteBuilder::from("timer:xsd-valid?period=3000&repeatCount=2")
        .route_id("xsd-valid")
        .set_body("<order><id>A1</id><amount>5</amount></order>")
        .log("Route 1: Validating XML order against XSD", LogLevel::Info)
        .validate(&xsd)
        .log("Route 1: XML is valid!", LogLevel::Info)
        .to("log:info?showBody=true")
        .build()?;

- id: xsd-valid
  from: timer:xsd-valid?period=3000&repeatCount=2
  steps:
    - set_body:
        value: "<order><id>A1</id><amount>5</amount></order>"
    - validate: "${body.contains('<order>')}"
    - to: log:info?showBody=true

The included route fires a timer twice. The .set_body(...) step sets the body to an XML order. The .validate(&xsd) step loads the XSD schema from the configured path and checks the body against it. The log step after the validator runs only when the body is valid. A second route in the example sends an invalid order. It pairs the validator with an error_handler, so the validation error is logged instead of crashing the process.

On a mismatch the validator returns an error. The error flows back through the same RouteErrorHandler boundary as any other step error. A route-level error_handler or a do-try catch block can recover it. Wrap the validator in a do-try block to keep the exchange in the pipeline after a failure. Use the bare validator to fail fast on bad input. The YAML validate step also accepts a predicate expression like ${body.contains('<order>')} for inline checks without a schema file.

The Validator differs from the Filter. The Validator stops the route on a failed check and surfaces the error. The Filter silently drops the exchange. Use the Validator for input validation and contract enforcement at a trust boundary. Use the Filter when an exchange is well-formed but not relevant.

Per ADR-0001, the validator compiles into a Service<Exchange> step. The schema engine runs inside the step. The processor contract is documented in camel-processor/CONTEXT.md.

The example source is at examples/validator.

Processing steps

Route steps that are not Enterprise Integration Patterns. They handle body materialization, method invocation, and flow control. These are route-building utilities, not named patterns from Hohpe and Woolf.

• Stream Cache — materialize a stream body into bytes so later steps can re-read it
• Bean — call a registered Rust function as a route step
• Stop — halt the pipeline without raising an error

For the route structure that hosts these steps, see Routes and pipelines.

Stream Cache

The Stream Cache step converts a Body::Stream into Body::Bytes up to a threshold. A stream drains on the first read. Steps that must read the body more than once cannot do so against a stream. Stream Cache reads the stream once, stores the bytes, and replaces the body. Every later step sees the same bytes.

RouteBuilder::from("timer:tick?period=1000")
    .route_id("stream-cache-demo")
    .stream_cache_default()
    .to("log:cached?showBody=true")
    .build()?;

- id: stream-cache-demo
  from: timer:tick?period=1000
  steps:
    - stream_cache: true
    - to: log:cached?showBody=true

The .stream_cache(threshold) call sets the maximum byte count the step stores in memory. Bodies smaller than the threshold become Body::Bytes. Bodies larger than the threshold stay as Body::Stream. The default threshold is 128 KB (DEFAULT_STREAM_CACHE_THRESHOLD). Call .stream_cache_default() to use this default.

In YAML, stream_cache: true enables caching with the default threshold. The form stream_cache: { threshold: N } sets a custom threshold in bytes.

Stream Cache is not an EIP. It is a pipeline utility that makes stream bodies safe for multi-read steps. Place a Stream Cache step before a Splitter that reads the body line by line. Place it before a Content Enricher that sends the original body and then inspects the response.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower pipeline. The stream body type is documented in camel-api/CONTEXT.md.

Bean

The Bean step calls a method on a registered bean. A bean is a named Rust object that lives in the bean registry. The step looks up the bean by name, calls the named method, and passes exchange data to it.

The #[bean_impl] and #[handler] macros turn a Rust impl block into a bean. Each #[handler] method becomes a callable entry point. The handler signature controls how the framework extracts data from the exchange. A parameter of type Order receives the deserialized body. The macro generates the binding code at compile time, so the binding is type-checked before the route runs.

RouteBuilder::from("timer:tick?period=1000")
    .route_id("bean-demo")
    .bean("orderProcessor", "handle")
    .to("log:processed?showBody=true")
    .build()?;

- id: bean-demo
  from: timer:tick?period=1000
  steps:
    - bean:
        name: orderProcessor
        method: handle
    - to: log:processed?showBody=true

The .bean("name", "method") call takes the bean name and method name as strings. The registry resolves the name to an instance. The method receives the exchange data and returns its result into the exchange body.

The Bean step is not an EIP. It is a Message Endpoint utility from the Hohpe and Woolf vocabulary. In Rust, the .process() closure serves the same role with a direct function reference. The Bean step exists primarily for the YAML DSL, where a closure is not available. A route that needs custom logic in YAML defines a bean, registers it, and calls it by name.

The Bean differs from Script. Script runs inline code declared in the route. Bean calls a pre-registered instance. A change to bean behavior needs a rebuild and re-registration. A change to script behavior needs only a route file edit.

Per ADR-0001, the step compiles into a Service<Exchange> in the Tower pipeline. The bean registry contract is documented in camel-bean/CONTEXT.md.

Stop

The Stop step halts the pipeline. No further steps run after a Stop. The exchange does not raise an error. The pipeline reports the outcome as PipelineOutcome::Stopped. This is a successful termination, not a failure.

RouteBuilder::from("timer:tick?period=1000")
    .route_id("stop-demo")
    .filter(|ex| ex.input.body.as_text().map(|t| t.contains("urgent")).unwrap_or(false))
    .to("log:urgent?showBody=true")
    .end_filter()
    .stop()
    .build()?;

- id: stop-demo
  from: timer:tick?period=1000
  steps:
    - filter:
        simple: "${body} contains 'urgent'"
        steps:
          - to: log:urgent?showBody=true
    - stop: true

The .stop() call adds a terminal step. When the exchange reaches this step, the pipeline returns PipelineOutcome::Stopped and the route stops processing that exchange. The exchange state is preserved as-is. The filter in the example shows the common pattern. Process exchanges that match a condition, then stop. Exchanges that fail the filter skip the filter block, reach the Stop step, and halt.

The Stop step is not an EIP. It is a flow control utility. It differs from a Message Filter. The Filter drops exchanges that fail the predicate. Stop halts every exchange that reaches it. A route that needs to stop conditionally places a filter or choice block before the Stop.

Stop is modeled as PipelineOutcome::Stopped, not as CamelError. This distinction matters for error handling. A stopped exchange does not trigger the route error handler. Place any log or retry step before the Stop step, not in an error handler.

Per ADR-0024, Stop is a CompiledStep::Stop variant. The executor converts it into PipelineOutcome::Stopped without invoking a Tower service. The outcome layer sits above Tower so the runtime can distinguish a stop from a failure.

Components

Components connect routes to external systems. Each Component owns a URI scheme and creates Endpoints that produce Consumers, Producers, or both. The vocabulary for Component, Endpoint, Consumer, and Producer lives in crates/components/CONTEXT.md.

Catalog

The table covers every crate under crates/components/. The contract crate camel-component-api defines the Component SPI and the Consumer, Producer, and Endpoint traits. It registers no URI scheme.

Direction

consumer marks an inbound Component. It starts a Consumer that submits Exchanges into the Route. producer marks an outbound Component. It creates a Producer that sends Exchanges to an external system. both means the Component supports either direction, one per Endpoint.

master wraps a delegate Consumer in a leadership gate. The bridge exposes inbound traffic only while this node holds the leadership lock (ADR-0035).

Narrative pages

• Timer and log. The smallest working route.
• File. Directory poller and disk writer.
• HTTP. Server Consumer and response handling.
• gRPC. Service consumer and producer.
• WebSocket and SOAP. Bidirectional WebSocket traffic and SOAP calls through the Java bridge.
• Kafka. Broker producer and consumer.
• JMS. Java bridge consumer and producer.
• MQTT. MQTT 3.1.1 broker producer and consumer.
• Redis. Datastore and pub/sub.
• Database. SQL access.
• SurrealDB. Multi-model database.
• OpenSearch. Search and indexing.
• LLM. Chat completions and embeddings.
• WASM. Sandboxed plugins with capability model.
• Cron. Scheduled message generation.
• Direct. Synchronous in-process routing.
• SEDA. Asynchronous staging between routes.
• ControlBus. Runtime control messages.
• Master. Leader-only route execution.
• Template. External template rendering.
• Validator. Schema validation.
• Exec. External process execution.
• Keycloak. OIDC auth and JWKS validation.
• XML transform. XSLT and JSON-XML conversion.
• Mock. Testing assertions.

Timer and log

The timer and log components form the smallest working route. Timer is a pure source. It fires Exchanges on a schedule and reads from no external system. Log is a pure sink. It formats Exchange state and writes the result through tracing. Together they exercise the source-to-sink path with no network and no extra dependencies.

The hello-world example wires both components and produces one Exchange per tick:

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    tracing_subscriber::fmt()
        .with_target(false) // Cleaner output
        .init();

    let mut ctx = CamelContext::builder().build().await.unwrap(); // allow-unwrap
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
        .route_id("hello-world") // Named routes
        .set_header("source", Value::String("timer".into()))
        .to("log:info?showHeaders=true&showCorrelationId=true") // Correlation ID
        .build()?;

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

    println!("Hello World example running. Press Ctrl+C to stop.");

    tokio::signal::ctrl_c().await.ok();
    ctx.stop().await?;
    Ok(())
}

routes:
  - id: hello-world
    from: "timer:tick?period=1000&repeatCount=5"
    steps:
      - set_header:
          key: source
          value: "timer"
      - to: "log:info?showHeaders=true&showCorrelationId=true"

Timer

timer:tick?period=1000&repeatCount=5 fires an Exchange every period milliseconds. The first tick fires immediately. The Consumer stops after repeatCount fires. Omit repeatCount to fire until the Route stops. Set repeatCount=0 and the timer fires never.

Timer is a consumer-only Component. Its Consumer submits one Exchange per tick. The Exchange body holds a short label. When includeMetadata=true, the Consumer sets four Exchange headers. They carry the timer name, the tick counter, the ISO-8601 fire time, and the epoch timestamp. Timer does not poll a directory and does not implement PollingConsumer. It is event-driven and runs inside the Route lifecycle.

Log

log:info?showHeaders=true&showCorrelationId=true writes the Exchange to the configured tracing level. The query parameters select what the Producer prints. showHeaders=true adds the Exchange header map. showCorrelationId=true prefixes the line with the correlation id, so several Routes can share one log output.

Log is a producer-only Component. Its Producer formats the Exchange body, headers, and correlation id, then returns the Exchange unchanged. The Producer composes with every other pipeline step (ADR-0001). Calling log: from a from: position returns an error at Endpoint creation.

The Producer exposes five levels: trace, debug, info, warn, error. Each level selects a tracing macro. Routes that handle sensitive data should set logMask=true. The option replaces the body and matching header values with a redaction marker.

Putting it together

The example registers both components before the Route starts. The Route flows from: timer:tick to to: log:info. Each Exchange passes through one Timer Consumer, one .set_header step, and one Log Producer. The result is one log line per tick for five ticks.

This pair is the recommended starting template. Once a route reads a header, transforms the body, or routes to more than one sink, replace log with a real sink and timer with a real source. The registration shape and the Endpoint URIs stay the same.

The timer URI grammar lives in the parent Components authority. The log contract surface lives in the camel-log CONTEXT. The example source is at examples/hello-world.

File

The file component is a directory poller for source routes and a disk writer for sink routes. The same crate covers both directions. The Consumer watches a directory for new or changed files. The Producer writes the Exchange body to disk under a chosen file name.

The file-pipeline example shows both directions with an upper-case transform and a dead-letter channel:

    let route = RouteBuilder::from(&format!(
        "file:{}?delete=true&initialDelay=0&delay=500&readTimeout=5000",
        input_path
    ))
    .route_id("file-pipeline-demo")
    .process(|mut exchange: camel_api::Exchange| {
        Box::pin(async move {
            if let Body::Text(text) = &exchange.input.body {
                let original_len = text.len();
                exchange.input.body = Body::Text(text.to_uppercase());
                exchange
                    .input
                    .set_header("original-length", Value::Number(original_len.into()));
            }
            Ok(exchange)
        })
    })
    .to(format!(
        "file:{}?fileExist=Override&writeTimeout=5000",
        output_path
    ))
    .to("log:pipeline?showHeaders=true&showBody=true&showCorrelationId=true")
    .error_handler(ErrorHandlerConfig::dead_letter_channel(
        "log:dead-letter?showBody=true&showHeaders=true&showCorrelationId=true",
    ))
    .build()?;

routes:
  - id: file-pipeline-demo
    from: "file:/tmp/rust-camel-pipeline/input?delete=true&initialDelay=0&delay=500&readTimeout=5000"
    error_handler:
      dead_letter_channel: "log:dead-letter?showBody=true&showHeaders=true&showCorrelationId=true"
    steps:
      - bean:
          name: uppercase-transform
          method: apply
      - to: "file:/tmp/rust-camel-pipeline/output?fileExist=Override&writeTimeout=5000"
      - to: "log:pipeline?showHeaders=true&showBody=true&showCorrelationId=true"

Source

file:{input}?delete=true&initialDelay=0&delay=500&readTimeout=5000 polls the input directory every delay milliseconds after an initialDelay wait. The Consumer submits one Exchange per detected file. The file body becomes the Exchange body. readTimeout=5000 bounds how long a poll waits for new content before it yields an empty result.

delete=true removes each file after a successful read. The deletion happens after the Route processes the Exchange. A route failure leaves the file in place, so the next poll retries it. Omit delete=true to keep every file after consumption. This fits audit pipelines and idempotent replay.

The Consumer is event-driven. It polls the directory on a schedule and pushes each file as an Exchange. It does not implement the on-demand PollingConsumer SPI. The consumer task starts with the Route and runs until the Route stops.

Sink

file:{output}?fileExist=Override&writeTimeout=5000 writes the Exchange body to the output directory. The fileExist parameter controls what happens when the target file already exists:

The Override and TryRename strategies route through one private atomic-write helper. The helper writes a temp file, then renames it over the target. The Fail strategy uses the OS-level atomic create_new(true) directly. The TryRename strategy requires an explicit tempPrefix. The Producer rejects path-traversal segments in fileName before any file operation. It also rejects a cross-filesystem rename (EXDEV) rather than falling back to a non-atomic copy.

fileName resolves from the CamelFileName Exchange header. A route can set this header in a process step to control the output name. The Producer creates missing parent directories on the path. Validation runs at Endpoint creation and at the Producer boundary. It fails closed (ADR-0033).

Set durable=true for crash safety. The Producer then fsyncs the temp file, performs the rename, then fsyncs the parent directory, in that order. This is crash-safe but slower. Leave durable=false for speed when crash safety is not required.

Pipeline shape

The example chains the source to a process step, then a sink, then log, then a dead-letter error handler. The from: Endpoint and the to: Endpoint share one CamelContext and resolve at Route start.

Use the file component when a directory on disk is the source or the sink. Use a different source when the data lives on a remote system that exposes a pull interface, such as SFTP. The file component polls a local path only. Use a different sink when the destination must be transactional, replicated, or ordered. File writes succeed per file with no cross-file consistency.

The atomic-write contract and the accepted fileExist values live in the camel-file CONTEXT. The example source is at examples/file-pipeline.

HTTP

The HTTP component covers both directions in one crate. The server side binds a TCP listener and consumes inbound requests. The client side produces outbound requests and forwards the response into the Route. The same http: scheme serves both. The direction follows the Endpoint position: from: for a server, to: for a client.

The http-server example shows the server direction with a process step that returns a JSON body:

    RouteBuilder::from("http://0.0.0.0:8080/health")
        .route_id("health-check")
        .to("log:health?showHeaders=true&showBody=true&showCorrelationId=true")
        .process(move |mut exchange| {
            let storage = Arc::clone(&storage);
            let rc = Arc::clone(&request_count);

            async move {
                let uptime = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0);

                exchange.input.body = Body::Json(serde_json::json!({
                    "status": "UP",
                    "service": "rust-camel-api",
                    "version": "1.0.0",
                    "uptime_seconds": uptime,
                    "requests_processed": rc.load(Ordering::Relaxed),
                    "users_count": storage.count(),
                    "timestamp": current_timestamp(),
                    "correlation_id": exchange.correlation_id,
                }));
                Ok(exchange)
            }
        })
        .build()
}

routes:
  - id: health-check
    from: "http://0.0.0.0:8080/health"
    steps:
      - to: "log:health?showHeaders=true&showBody=true&showCorrelationId=true"
      - bean:
          name: health-handler
          method: build_status

The http-client example shows the client direction. A timer drives a GET request to a remote service, and a log step records the response:

    let route = RouteBuilder::from("timer:http-poll?period=5000&repeatCount=3")
        .route_id("http-client")
        .process(|mut exchange| {
            Box::pin(async move {
                exchange.input.body = Body::Empty;
                exchange
                    .input
                    .set_header("X-Request-Id", Value::String(uuid_header()));
                Ok(exchange)
            })
        })
        .to("https://httpbin.org/get?source=rust-camel&allowInternal=false")
        .process(|exchange| {
            Box::pin(async move {
                let body_str = match &exchange.input.body {
                    Body::Text(s) => Some(s.as_str()),
                    Body::Bytes(b) => std::str::from_utf8(b).ok(),
                    _ => None,
                };
                if let Some(text) = body_str
                    && let Ok(json) = serde_json::from_str::<serde_json::Value>(text)
                {
                    println!(
                        "Response status: {:?}",
                        exchange.input.header("CamelHttpResponseCode")
                    );
                    if let Some(url) = json.get("url") {
                        println!("Response from: {}", url);
                    }
                }
                Ok(exchange)
            })
        })
        .to("log:http-response?showHeaders=true&showBody=true&showCorrelationId=true")
        .error_handler(
            ErrorHandlerConfig::dead_letter_channel("log:http-dlc?showBody=true")
                .on_exception(|_| true)
                .retry(2)
                .build(),
        )
        .build()?;

routes:
  - id: http-client
    from: "timer:http-poll?period=5000&repeatCount=3"
    error_handler:
      dead_letter_channel: "log:http-dlc?showBody=true"
      retry:
        max_attempts: 2
    steps:
      - bean:
          name: request-id-setter
          method: apply
      - to: "https://httpbin.org/get?source=rust-camel&allowInternal=false"
      - bean:
          name: response-printer
          method: apply
      - to: "log:http-response?showHeaders=true&showBody=true&showCorrelationId=true"

Server

http://0.0.0.0:8080/health is a Consumer Endpoint. The Runtime binds a TCP listener on the address and port from the URI, then dispatches each inbound request to the Route. The Consumer submits one Exchange per request. The Exchange body carries the request payload. The Exchange headers carry the request headers, the path, the query string, and the request metadata.

The example binds 0.0.0.0 to listen on every interface. Use 127.0.0.1 when the server must stay on the loopback. The component rejects a partial server TLS configuration that supplies only a certificate or only a key. Operators who terminate TLS at a separate proxy need no server TLS in the URI.

The Consumer bounds resource use. It enforces a 2 MiB default request-body limit, a read timeout, and an in-flight request semaphore. The reply finaliser maps the Exchange body shape to a response. A Body::Json value serializes to JSON. A Body::Stream value streams the response. The CamelHttpResponseCode header sets the HTTP status. A missing header produces 200 OK.

Client

to: http://example.org/api/data is a Producer Endpoint. The Producer builds an outbound request, sends it through the configured transport, and returns the response as the new Exchange body. The Producer follows redirects, applies the configured NetworkRetryPolicy, and surfaces transport failures as CamelError.

The component validates each outbound URL and each redirect hop. allow_internal=false rejects internal addresses by default. The Producer pins DNS resolution to validated addresses to prevent DNS rebinding. Cross-origin redirects drop the Authorization and Cookie headers. The Producer has a 10 MiB default response-body limit.

TlsConfig verifies peer certificates by default. An operator can opt out with insecure=true or verify_peer=false. The opt-out emits a warning. The component forbids cleartext HTTP to public addresses even with allow_internal=true.

Direction choice

A route that serves an API uses HTTP as a source. A route that calls another service uses HTTP as a sink. A route that does both needs two Endpoints, one in each position. The second Endpoint belongs to a second Route. The component supports both directions because the same crate owns the connection plumbing. Each Endpoint still picks one direction at creation time.

The inbound and outbound contract surface lives in the camel-http CONTEXT. The diagnostic endpoints (/healthz, /readyz, /metrics) live in the Operations authority. The example sources are at examples/http-server and examples/http-client.

gRPC

The gRPC component produces and consumes gRPC with runtime proto resolution. No compile-time code generation is required. The component resolves .proto files at runtime through camel-proto-compiler and prost-reflect. It supports unary, server-streaming, client-streaming, and bidirectional streaming. The mode is auto-detected from the proto method descriptor.

The grpc-example wires a consumer on port 50051 and a timer-driven producer:

    let consumer_route = RouteBuilder::from(&format!(
        "grpc://0.0.0.0:50051/helloworld.Greeter/SayHello?protoFile={}",
        proto_path
    ))
    .set_body(Body::Json(
        serde_json::json!({"message": "Hello from consumer!"}),
    ))
    .to("log:grpc-consumer?showBody=true")
    .build()?;

routes:
  - id: grpc-consumer
    from: "grpc://0.0.0.0:50051/helloworld.Greeter/SayHello?protoFile=helloworld.proto&transport=plaintext"
    steps:
      - set_body:
          value:
            message: Hello from consumer!
      - to: "log:grpc-consumer?showBody=true"

    let producer_route = RouteBuilder::from("timer:grpc-tick?period=3000&repeatCount=3")
        .set_body(Body::Json(serde_json::json!({"name": "World"})))
        .to(format!(
            "grpc://127.0.0.1:50051/helloworld.Greeter/SayHello?protoFile={}",
            proto_path
        ))
        .to("log:grpc-response?showBody=true")
        .build()?;

routes:
  - id: grpc-producer
    from: "timer:grpc-tick?period=3000&repeatCount=3"
    steps:
      - set_body:
          value:
            name: World
      - to: "grpc://127.0.0.1:50051/helloworld.Greeter/SayHello?protoFile=helloworld.proto&transport=plaintext"
      - to: "log:grpc-response?showBody=true"

URI

grpc://<host>:<port>/<package>.<Service>/<Method>?protoFile=<path>&transport=<mode>

Consumer

grpc://0.0.0.0:50051/helloworld.Greeter/SayHello?protoFile=helloworld.proto&transport=plaintext binds an HTTP/2 listener. The Consumer dispatches each inbound gRPC request to the Route. The Exchange body carries the decoded protobuf message as JSON. The Exchange headers carry gRPC metadata.

Multiple GrpcConsumers on the same (host, port) share one HTTP/2 server. Each consumer registers dispatch by URI path. The shared-server registry refuses to mix TLS and plaintext on one listener.

The Consumer supports four RPC modes. It auto-detects the mode from the proto method descriptor. The same Consumer handles unary, server-streaming, client-streaming, and bidirectional calls without configuration changes.

Producer

grpc://127.0.0.1:50051/helloworld.Greeter/SayHello?protoFile=helloworld.proto&transport=plaintext sends the Exchange body as a gRPC call. The Producer holds a lazy pool of connections. It reports endpoint health through RuntimeObservability.

The Producer requires transport=plaintext or transport=tls in the URI (ADR-0033). The legacy tls=true and tls=false keys are rejected. Under tls, the endpoint URL is rewritten to https://. The insecure_skip_verify=true option hard-errors. The component fails closed on an incomplete mTLS identity.

Security

The component enforces two security layers (ADR-0010, ADR-0032):

1. Authentication. The server handler parses Authorization: Bearer <token> and calls TokenAuthenticator::authenticate_bearer. Missing or invalid credentials return Status::unauthenticated.
2. Authorization. The Consumer calls SecurityPolicy::evaluate before pipeline dispatch in all four RPC modes. Denied returns Status::permission_denied. Unknown future decisions fail closed.

Transport setup also fails closed (ADR-0033). Every Endpoint declares transport=plaintext or transport=tls. The component rejects insecure_skip_verify=true, an incomplete mTLS identity, and a TLS/plaintext mismatch on a shared listener.

Streaming

Routes that send streaming responses use GrpcStreamObserver. The observer exposes three methods: on_next, on_error, and on_completed. The route calls these methods to push response messages onto the gRPC stream.

Reference: gRPC crate CONTEXT. Example source: examples/grpc-example.

WebSocket and SOAP (CXF)

The WebSocket component (ws, wss) serves and connects to real-time bidirectional endpoints. The CXF component (cxf) calls and hosts SOAP services through a supervised Java bridge. Both speak plain URI syntax. No code generation. No SOAP engine in the Rust data plane.

The ws-server example wires a chat broadcast and a per-connection echo against the same port:

use camel_api::{CamelError, Value};
use camel_builder::{RouteBuilder, StepAccumulator};
use camel_component_ws::WsComponent;
use camel_core::context::CamelContext;

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    let mut ctx = CamelContext::builder().build().await.unwrap();
    ctx.register_component(WsComponent::new());

    // Echo: CamelWsConnectionKey arrives on the inbound message; the producer
    // targets the same key on the same path, so the reply reaches the sender.
    let echo_route = RouteBuilder::from("ws://0.0.0.0:9000/echo")
        .route_id("ws-echo")
        .to("ws://0.0.0.0:9000/echo")
        .build()?;

    // Chat: CamelWsSendToAll=true tells the producer to fan out to every
    // local client connected to /chat.
    let chat_route = RouteBuilder::from("ws://0.0.0.0:9000/chat")
        .route_id("ws-chat")
        .set_header("CamelWsSendToAll", Value::Bool(true))
        .to("ws://0.0.0.0:9000/chat")
        .build()?;

    ctx.add_route_definition(echo_route).await?;
    ctx.add_route_definition(chat_route).await?;
    ctx.start().await?;
    Ok(())
}

routes:
  - id: ws-echo
    from: "ws://0.0.0.0:9000/echo"
    steps:
      - to: "ws://0.0.0.0:9000/echo"
  - id: ws-chat
    from: "ws://0.0.0.0:9000/chat"
    steps:
      - set_header:
          name: "CamelWsSendToAll"
          value: "true"
      - to: "ws://0.0.0.0:9000/chat"

WebSocket URI

ws://host:port/path[?options]
wss://host:port/path[?options]

TLS uses rustls. No OpenSSL dependency.

WebSocket consumer

ws://0.0.0.0:9000/chat accepts inbound upgrade requests and submits one Exchange per received frame. The Consumer sets three headers on every inbound message:

A ServerRegistry keys servers by port. The first Consumer to register on a port fixes the host and TLS mode. A later registration with the other TLS mode fails (Server on port N already running with different TLS mode). Multiple Consumers on the same port share one server and register independent paths. Consumer shutdown removes the path, the security policy, and the connection registry. The server stays alive until the process exits.

The upgrade handler (dispatch_handler) checks Origin against allowOrigin before any auth step. When a SecurityContext is attached, the handler fails closed on missing credentials, denied policy decisions, and future AuthorizationDecision variants. Query-token values are redacted before logging (ADR-0051). TLS certificate and key paths in WsEndpointConfig::fmt are redacted as well. The crate applies a stricter diagnostic policy than ADR-0051 because the cert and key paths sit on the same boundary.

WebSocket producer

The Producer has two modes. The mode is selected by the inbound Exchange and the URI:

• Outbound client mode when no local Consumer matches the URI. The Producer opens a Tokio-tungstenite connection, sends the body, and reads the first reply.
• Server-send mode when a local Consumer matches the URI. The Producer never opens a new socket. It writes to the local connection registry.

The mode picker inspects three signals: the CamelWsSendToAll header, the CamelWsConnectionKey header, and the presence of a local Consumer on (host, port, path). Any one of them selects server-send mode. The producer then targets the keys in CamelWsConnectionKey (comma-separated) or, with CamelWsSendToAll=true, every active connection on the path.

The body becomes a Text frame by default. Set CamelWsMessageType to binary for a Binary frame. The URI parameter binaryPayload=true makes binary the default for that endpoint.

The client mode reconnects on transient failures with a shared NetworkRetryPolicy. The producer also surfaces backpressure: when a server-send channel is full, it sets CamelWsDeliveryDropped and returns Err on the next poll_ready call.

WebSocket security

Auth rides on the per-route SecurityContext (see Authentication and authorization). When a path has a context, the upgrade handler authenticates the bearer token and evaluates the route's SecurityPolicy before completing the upgrade. Failed auth returns 401. Failed policy returns 403. Policy evaluation errors return 500 and increment the e:ws:policy-eval metric (ADR-0012 class e). The health pin becomes the operator signal for bind failures (ADR-0012 class g).

CXF URI

cxf://http://host:port/path?wsdl=file.wsdl&service={ns}Name&port={ns}Port[&operation=opName][&profile=profile_name][&timeout_ms=N][&mtom_enabled=true|false]

The Rust component treats the SOAP envelope and WSDL as opaque bytes. No XML parser runs in this crate. The supervised Java bridge owns parsing, DTD handling, and entity resolution. The XXE boundary lives in the bridge process (ADR-0032).

The cxf-example ships a producer that fires every ten seconds and a consumer that returns a fixed response:

routes:
  - id: cxf-producer-hello
    from:
      uri: "timer:hello-tick?period=10000"
    steps:
      - set-body:
          constant: "<sayHello><name>World</name></sayHello>"
      - to: "cxf://http://localhost:8080/hello?wsdl=wsdl/hello.wsdl&service={http://example.com/hello}HelloService&port={http://example.com/hello}HelloPort&operation=sayHello"
      - log: "SOAP response: ${body}"

routes:
  - id: cxf-consumer-hello
    from:
      uri: "cxf://http://0.0.0.0:9090/hello?wsdl=wsdl/hello.wsdl&service={http://example.com/hello}HelloService&port={http://example.com/hello}HelloPort"
    steps:
      - log: "Received SOAP request: ${body}"
      - set-body:
          simple: "<sayHelloResponse><message>Hello from rust-camel!</message></sayHelloResponse>"

The example ships with Camel.toml that declares the profile:

[components.cxf]
version = "0.8.1"

[[components.cxf.profiles]]
wsdl_path = "wsdl/hello.wsdl"
service_name = "{http://example.com/hello}HelloService"
port_name = "{http://example.com/hello}HelloPort"

CXF bridge

The component spawns a single GraalVM native Java bridge process on first use. The Rust side calls ensure_binary_for_spec to download (or reuse) the bridge binary, then starts it with CXF_PROFILES and CXF_PROFILE_<NAME>_* environment variables. The bridge prints a JSON readiness line on stdout; the Rust side reads the ephemeral gRPC port and the health endpoint URL from that line.

Profile names must match [a-z0-9_]+. The validator rejects uppercase, hyphens, and spaces at config load time. Each profile owns its own WSDL, service, port, and optional keystore and truststore. Multiple profiles share one bridge process.

A background health monitor probes the bridge on a configurable interval (default 5s). On a failed health check the monitor moves the slot to Degraded and triggers a restart with exponential backoff (capped at 30s). After 10 failed restart attempts the monitor transitions the slot to permanent Degraded. The route startup fails fast when the initial probe returns Degraded or Stopped.

On shutdown the order matters. Call pool.begin_shutdown() before ctx.stop() so the health monitor does not race a restart against shutdown. The CLI handles this for you with Ctrl+C.

CXF producer

cxf://http://host:port/path?... sends the Exchange body as a SOAP request. The body bytes go over gRPC with the configured security_profile selector. The bridge resolves the profile, signs the envelope with the profile's certificate when WS-Security is configured, and POSTs the request to the SOAP target. The response bytes come back as the Exchange body.

The profile query parameter is required. The endpoint resolver looks up the profile in CxfBridgePool::configured_profiles; an unknown name fails endpoint creation with unknown profile 'X'. The operation parameter is optional when the WSDL declares a single operation. Use the CamelCxfOperation header to override the operation per Exchange.

CXF consumer

cxf://http://0.0.0.0:9090/... hosts a SOAP endpoint. The bridge process publishes the endpoint under <base>/<profile_name> over Vert.x. Inbound SOAP requests arrive on the gRPC stream as ConsumerRequest messages. The Consumer builds an Exchange with the payload bytes as the body and the headers from the table below.

The response body is what the route writes to exchange.input.body (or exchange.output.body if the route set it). The bridge wraps the bytes in a SOAP envelope and signs with the profile's certificates.

A failed route handler returns a soap:Server fault with the error string. A response-marshalling failure (for example a Stream body, which CXF does not support) also returns a soap:Server fault, increments b-prime:cxf:response-marshalling, and logs at error! (ADR-0012 class b').

The inbound Exchange carries these headers from the bridge:

Other HTTP headers from the request land in the Exchange header map.

CXF security

Credentials flow through per-profile environment variables. CxfProfileEnvVars wraps keystore, truststore, and signature passwords in Redacted while it builds the child-process configuration. gRPC requests carry only the security_profile selector. They do not carry password bytes. CxfSecurityFields::fmt prints <redacted> for every password field. The redaction policy follows ADR-0051.

The XML parser boundary lives in the Java bridge. The Rust component does not run an XML parser. Audit parser hardening in the bridge process. WSDL files also live in the bridge. The Rust side reads them as paths and forwards them to the bridge over the environment.

[CXF-014]: the MTOM multipart body is not yet fully implemented. The flag sets headers but the binary part is not encoded.

Reference: WebSocket crate CONTEXT, CXF crate CONTEXT, ADR-0032 trust boundary, ADR-0012 log levels, ADR-0051 credential redaction. Example source: examples/ws-server, examples/cxf-example.

Kafka

The Kafka component produces to and consumes from Apache Kafka topics. One crate covers both directions. The Consumer subscribes to topics and submits one Exchange per record. The Producer publishes the Exchange body to a topic.

The kafka-example wires a timer-driven producer and a log consumer against a testcontainers broker:

    let producer_route = RouteBuilder::from("timer:tick?period=3000")
        .route_id("kafka-producer")
        .set_body(Value::String(
            r#"{"event":"heartbeat","source":"rust-camel"}"#.to_string(),
        ))
        .to(format!(
            "kafka:{topic}?brokers={brokers}&acks=all",
            topic = TOPIC,
            brokers = brokers
        ))
        .build()?;

routes:
  - id: kafka-producer
    from: "timer:tick?period=3000"
    steps:
      - set_body: '{"event":"heartbeat","source":"rust-camel"}'
      - to: "kafka:orders?brokers=127.0.0.1:9092&acks=all"

    let consumer_route = RouteBuilder::from(&format!(
        "kafka:{topic}?brokers={brokers}&groupId=example-group&autoOffsetReset=earliest",
        topic = TOPIC,
        brokers = brokers
    ))
    .route_id("kafka-consumer")
    .to("log:info?showHeaders=true")
    .build()?;

routes:
  - id: kafka-consumer
    from: "kafka:orders?brokers=127.0.0.1:9092&groupId=example-group&autoOffsetReset=earliest"
    steps:
      - to: "log:info?showHeaders=true"

URI

kafka:<topic>?brokers=<host:port>[&groupId=<group>][&autoOffsetReset=<policy>][&partitionAssignmentStrategy=<strategy>][&acks=<level>][&securityProtocol=<protocol>]

Consumer

kafka:<topic>?brokers=localhost:9092&groupId=my-group subscribes to a topic. The Consumer submits one Exchange per record. The Exchange body carries the record value. The headers carry the topic, partition, offset, key, and timestamp (CamelKafkaTopic, CamelKafkaPartition, CamelKafkaOffset, CamelKafkaKey, CamelKafkaTimestamp).

The groupId coordinates consumption across instances. Consumers that share a group ID split the topic partitions. The partitionAssignmentStrategy picks how the broker assigns partitions across members: range (default), roundRobin, or cooperativeSticky. The autoOffsetReset policy picks the start position when the group has no committed offset: latest (default), earliest, or none.

Offset commit has two modes. Auto-commit is the default. The Consumer commits offsets on its auto-commit interval. Set allowManualCommit=true to commit from the route instead. The route reads a KafkaManualCommit handle from the kafka.manual_commit exchange property and calls commit_async() after it processes the record.

The Consumer uses the standard push model. It does not implement PollingConsumer. The consumer task starts with the Route and runs until the Route stops or reconnect attempts exhaust.

Producer

kafka:<topic>?brokers=localhost:9092&acks=all sends the Exchange body to a topic. The acks parameter controls durability. all waits for every in-sync replica. 1 waits for the leader only. 0 fires and forgets. On success the Producer returns the Exchange unchanged and writes delivery metadata to the CamelKafkaRecordMetadata header.

A send failure returns Err. The pipeline catches it and the route ErrorHandler owns the operational signal.

Security

The component supports four security protocols:

• PLAINTEXT (default). No encryption. The component warns at startup.
• SSL. TLS encryption. Gated behind the ssl or ssl-vendored cargo feature.
• SASL_PLAINTEXT. SASL authentication without encryption. Gated behind the sasl feature.
• SASL_SSL. SASL authentication with TLS. Requires both the sasl and ssl features.

The component redacts SASL and SSL passwords to [REDACTED] in Debug output. A missing feature gate stops startup with the required cargo add command.

Error handling

The Consumer logs at error! for auto-commit failures, manual-commit handler failures, and exhausted reconnect attempts. ADR-0012 classifies these as outside-contract (b') and system-broken (c). The Producer logs send failures at warn!. The route handler owns these failures (ADR-0012 category a).

Reference: Kafka crate CONTEXT. Example source: examples/kafka-example.

JMS

The JMS component publishes to and consumes from JMS brokers through a Java bridge process. It supports ActiveMQ Classic and ActiveMQ Artemis. One crate covers both directions. The Consumer subscribes to a destination and submits one Exchange per message. The Producer publishes the Exchange body to a destination.

The component does not implement JMS in Rust. It delegates protocol work to a native Java bridge binary over gRPC. The bridge is downloaded once and cached.

Schemes

Three schemes share one bridge pool:

The jms: scheme requires an explicit destination type. jms:orders returns an error. Use jms:queue:orders or jms:topic:orders.

The activemq: and artemis: schemes set the broker type at the URI level. They override any broker_type declared in the broker config. Use jms: when you want the broker type to come from configuration.

Example

The jms-example wires a timer-driven producer and a log consumer against a testcontainers broker:

    let producer_route = RouteBuilder::from("timer:tick?period=3000")
        .route_id("jms-producer")
        .set_body(Value::String(
            r#"{"event":"order","source":"rust-camel"}"#.to_string(),
        ))
        .to("activemq:queue:orders")
        .build()?;

routes:
  - id: jms-producer
    from: "timer:tick?period=3000"
    steps:
      - set_body: '{"event":"order","source":"rust-camel"}'
      - to: "activemq:queue:orders"

    let consumer_route = RouteBuilder::from("activemq:orders")
        .route_id("jms-consumer")
        .to("log:info?showHeaders=true")
        .build()?;

routes:
  - id: jms-consumer
    from: "activemq:orders"
    steps:
      - to: "log:info?showHeaders=true"

URI

jms:queue:<name>[?param=value&...]
jms:topic:<name>[?param=value&...]
activemq:queue:<name>[?param=value&...]
activemq:topic:<name>[?param=value&...]
artemis:queue:<name>[?param=value&...]
artemis:topic:<name>[?param=value&...]
activemq:<name>      # shorthand, defaults to queue
artemis:<name>       # shorthand, defaults to queue

Credentials and the broker URL are not URI parameters. They live in the [components.jms.brokers.<name>] table.

Broker configuration

Brokers are declared in Camel.toml. The component creates one Java bridge process per broker. The bridge pool admits at most max_bridges (default 8) bridges concurrently.

[default.components.jms]
default_broker = "main"

[default.components.jms.brokers.main]
broker_url  = "tcp://localhost:61616"
broker_type = "activemq"   # "activemq" | "artemis"
username    = "admin"      # optional
password    = "admin"      # optional

The bridge binary downloads on first use. It comes from a configured release URL and is cached at ~/.cache/rust-camel/jms-bridge/. The download is SHA256-verified. Set CAMEL_JMS_BRIDGE_BINARY_PATH to point at a local build for development.

Consumer

activemq:queue:orders subscribes to a destination. The Consumer submits one Exchange per inbound JMS message. The Exchange body carries the message payload. With mapJmsHeaders=true (the default), the headers carry JMSMessageID, JMSCorrelationID, JMSTimestamp, JMSDestination, and JMSPriority.

The body is typed from the JMS content_type. text/* becomes Body::Text. application/json becomes Body::Json when valid JSON. Binary content becomes Body::Bytes. The consumer pre-flights the bridge slot before starting. A missing or degraded bridge fails fast with JMS bridge not available.

The concurrentConsumers parameter spawns N parallel consumer tasks on the same destination. Each task subscribes independently and submits Exchanges into the shared route pipeline. messageSelector filters messages at the broker with a SQL-92 expression.

Producer

activemq:queue:orders sends the Exchange body to a destination. The content type is inferred from the body. Body::Text sends text/plain. Body::Json sends application/json. Body::Xml sends text/xml. An explicit Content-Type header wins over inference.

The Producer uses a semaphore for backpressure. The default concurrency limit is 128 in-flight sends. When the limit is reached, poll_ready returns Pending. ADR-0024 classifies a closed semaphore as ConsumerStopping.

A successful send returns the Exchange unchanged. The JMSMessageID header carries the broker-assigned message ID. A send failure on a gRPC transport error refreshes the channel. The original send is not retried. A retry on a non-idempotent write would cause duplicates. The caller decides whether to retry.

Java bridge

The component does not speak JMS. A native Java bridge process handles the JMS protocol and the TCP connection. The component talks to the bridge over gRPC. The bridge binary is a jlink image. The host does not need a Java runtime.

The bridge pool assigns one bridge per broker. The first send or consume starts the bridge. The bridge's ephemeral gRPC port comes from its stdout. A health monitor pings the bridge every healthCheckIntervalMs (default 5s). A failed health check moves the slot to Degraded, then Restarting. Restarts use exponential backoff capped at 120 seconds. After 10 failed restart attempts the slot stays Degraded.

The Consumer observes bridge state through a watch channel. A pending bridge returns Pending from poll_ready. A degraded bridge returns Err with the reason. The producer waits for the bridge to become Ready before sending.

Trust boundary

Per ADR-0032, incoming JMS headers, bodies, correlation IDs, and destinations enter exchange.input without validation. The route is responsible for validation when data crosses into a control action, a resource decision, or an executable sink.

Credentials cross the gRPC boundary in plain text for the username and through a Redacted wrapper for the password. BrokerConfig redacts the password in Debug output. BridgeSlot omits its credentials field. redact_url strips user information from URLs before logging. The audit command rg '#\[derive.*Debug' crates/components/camel-jms/src/ checks that no type holding a password derives Debug.

Error handling

ADR-0007 governs Consumer shutdown. JmsConsumer::stop cancels the CancellationToken and waits up to 5 seconds for the consumer tasks to finish. Tasks that do not exit in that window are aborted. An in-flight ConsumerContext::send completes before the loop checks cancellation. The Consumer does not restart itself. Route supervision owns restart.

ADR-0012 classifies log sites as outside-contract or system-broken. A consumer-side ctx.send failure increments the b-prime:jms:consumer-send metric and logs at error!. A bridge restart that exhausts the attempt cap logs at error! and leaves the slot in Degraded.

Limitations

• The bridge uses AUTO_ACKNOWLEDGE. Messages are acknowledged on delivery, not after processing. A failed route cannot request redelivery.
• Durable topic subscribers are not supported.
• IBM MQ is not supported.
• The InOut exchange pattern and Session transaction mode log a warning and fall back to the default.

Reference: JMS crate CONTEXT. Example source: examples/jms-example.

MQTT

The MQTT component publishes to and consumes from MQTT 3.1.1 brokers. The Consumer subscribes to topic filters and submits one Exchange per incoming publish. The Producer is a Tower Service<Exchange> that publishes the Exchange body to a topic.

The mqtt-example wires a timer-driven producer and a log consumer against a Mosquitto broker started by testcontainers:

    let producer = RouteBuilder::from("timer:tick?period=3000&repeatCount=2")
        .route_id("mqtt-producer")
        .set_body("hello-mqtt")
        .to("mqtt://test/sensors/temp")
        .build()?;

routes:
  - id: mqtt-producer
    from: "timer:tick?period=3000&repeatCount=2"
    steps:
      - set_body: "hello-mqtt"
      - to: "mqtt://test/sensors/temp"

    let consumer = RouteBuilder::from("mqtt://test/sensors/#")
        .route_id("mqtt-consumer")
        .to("log:info?showHeaders=true")
        .build()?;

routes:
  - id: mqtt-consumer
    from: "mqtt://test/sensors/#"
    steps:
      - to: "log:info?showHeaders=true"

URI

mqtt://<broker_name>[/<topic>][?query]
mqtts://<broker_name>[/<topic>][?query]

<broker_name> is a logical key, not a host:port. The component resolves it to a MqttBrokerConfig declared in Camel.toml under [components.mqtt.brokers.<name>]. The path segment becomes the default subscription filter for the Consumer and the default publish topic for the Producer. Use the topics query parameter for multi-filter subscriptions.

Invalid qos or ackMode values fail endpoint creation with CamelError::Config.

Consumer

mqtt://<broker>/sensors/# subscribes to a topic filter. The Consumer opens a TCP connection to the broker, subscribes, and feeds each incoming publish into the route as an Exchange. The Exchange body carries the MQTT payload. The headers carry CamelMqttTopic, CamelMqttQos, CamelMqttRetained, CamelMqttDuplicate, CamelMqttClientId, and CamelMqttPacketId (the last only for QoS 1 and 2).

Each route Consumer opens its own TCP connection. v1 has no shared connection pool (ADR-0027). Account for the per-route connection when you size your broker.

Manual ack with QoS 1 or 2 requires cleanSession=false. With cleanSession=true the broker discards session state on reconnect and unacknowledged messages cannot be redelivered. Validation rejects the unsafe combination at endpoint creation. The ack decision uses the received packet QoS, not the subscription QoS. A subscription at QoS 1 can still receive QoS 0 messages, and those messages must never be manually acked.

Producer

mqtt://<broker>/sensors/temp publishes the Exchange body to a topic. Each route endpoint creates a Tower Service<Exchange> Producer. Each Producer opens its own TCP connection to the broker (ADR-0027).

The body becomes the MQTT payload. The URI path sets the default publish topic and QoS. The headers CamelMqttTopic, CamelMqttQos, and CamelMqttRetain override the defaults for that exchange. CamelMqttTopic must not contain + or #.

Connection retries use the shared NetworkRetryPolicy with exponential backoff and jitter. Every backoff sleep is cancellation-aware and stops when the route shuts down. The driver loop logs retried connection errors at warn!. It has no runtime handle so it cannot call error! with a replacement signal.

Configuration

Brokers live in Camel.toml under [components.mqtt.brokers]. The URI references the broker by name. Credentials stay in the config file, never in the route:

[default.components.mqtt]
client_id_prefix = "camel"

[default.components.mqtt.brokers.my-broker]
url = "mqtt://mqtt.example.com:1883"
username = "app-user"
password = "app-secret"

The url field accepts mqtt:// (plain TCP) or mqtts:// (TLS). mqtts:// requires the tls cargo feature. The default features include TLS, so connections use rustls out of the box. To compile without TLS, disable default features:

camel-component-mqtt = { version = "0.20", default-features = false }

The component redacts the broker password in Debug output. mTLS (client certificate authentication) is not yet supported in v1.

Connection lifecycle

Every Consumer and every Producer opens one TCP connection to the broker. The component uses no shared connection in v1. The connection carries a 10-second timeout. Unreachable brokers fail fast instead of hanging.

The auto-generated client_id follows the pattern {prefix}-{route_id}-{hash6} and truncates to 23 bytes (the MQTT 3.1.1 portable maximum). The hash input is the full endpoint URI for producers and the broker name plus subscription list for consumers. Set the clientId URI parameter to override.

Reference: MQTT crate CONTEXT. Architecture decisions: ADR-0027. Example source: examples/mqtt-example.

Redis

The Redis component executes Redis commands and subscribes to Redis channels. One crate covers both directions. The Producer sends the Exchange body to Redis as a command argument. The Consumer subscribes to Pub/Sub channels or blocks on a list key. The redis URI scheme uses plaintext. The rediss URI scheme uses TLS.

The redis-example wires a string producer, a Pub/Sub consumer, a queue consumer, and a Pub/Sub producer against a testcontainers Redis instance:

use camel_builder::{RouteBuilder, StepAccumulator};
use camel_component_redis::RedisComponent;

ctx.register_component("redis", Box::new(RedisComponent::new()));

// Producer: timer writes a key every 3s
let string_producer = RouteBuilder::from("timer:tick?period=3000&repeatCount=3")
    .route_id("redis-string-producer")
    .set_header("CamelRedis.Key", Value::String("greeting".into()))
    .set_header(
        "CamelRedis.Value",
        Value::String("hello from rust-camel!".into()),
    )
    .to("redis://127.0.0.1:6379?command=SET")
    .to("log:info?showHeaders=true")
    .build()?;

routes:
  - id: redis-string-producer
    from: "timer:tick?period=3000&repeatCount=3"
    steps:
      - set_header:
          CamelRedis.Key: "greeting"
      - set_header:
          CamelRedis.Value: "hello from rust-camel!"
      - to: "redis://127.0.0.1:6379?command=SET"
      - to: "log:info?showHeaders=true"

// Consumer: BRPOP blocks on a list key, one Exchange per popped item
let queue_consumer = RouteBuilder::from(
    "redis://127.0.0.1:6379?command=BRPOP&key=demo-queue&timeout=2",
)
.route_id("redis-queue-consumer")
.to("log:info?showAll=true")
.build()?;

// Consumer: SUBSCRIBE receives published messages as Exchanges
let pubsub_consumer = RouteBuilder::from(
    "redis://127.0.0.1:6379?command=SUBSCRIBE&channels=demo-channel",
)
.route_id("redis-pubsub-consumer")
.to("log:info?showAll=true")
.build()?;

routes:
  - id: redis-queue-consumer
    from: "redis://127.0.0.1:6379?command=BRPOP&key=demo-queue&timeout=2"
    steps:
      - to: "log:info?showAll=true"
  - id: redis-pubsub-consumer
    from: "redis://127.0.0.1:6379?command=SUBSCRIBE&channels=demo-channel"
    steps:
      - to: "log:info?showAll=true"

URI

redis://host:port?command=<cmd>[&key=<key>][&channels=<list>][&timeout=<secs>][&password=<pwd>][&db=<n>][&ssl=<bool>]

The command parameter picks the Redis command at Endpoint creation. Exchange data never becomes a command name. Dynamic values like keys, fields, values, channels, and scores cross the trust boundary as length-prefixed Redis protocol arguments. Argument contents cannot inject a second command or change the selected command (CONTEXT "Trust boundary"). Missing required headers return CamelError.

Commands

The component exposes 80+ commands across eight groups. The enum is exhaustive: an unknown command fails URI parsing with CamelError::InvalidUri. The component does not expose EVAL, EVALSHA, or script-loading commands. Script injection through the public surface is not possible.

Producer

redis://host:port?command=GET sends the Exchange body to Redis. The Producer holds a single multiplexed connection per Endpoint. The connection opens lazily on the first call and stays open for the route lifetime.

The command parameter picks one of the 80+ commands listed above. The Exchange body, headers, and the URI parameters supply the command arguments. Different commands read different headers. For example, HSET reads CamelRedis.Key and CamelRedis.Value. LRANGE reads CamelRedis.Start and CamelRedis.End. Missing required headers return CamelError. A send failure returns Err to the route ErrorHandler.

The Producer is a Tower Service<Exchange>. It composes with any pipeline step and reports per-route metrics.

Consumer

redis://host:port?command=SUBSCRIBE&channels=foo,bar subscribes to one or more Pub/Sub channels. The Consumer submits one Exchange per published message. The CamelRedis.Channel header carries the channel name. CamelRedis.Pattern carries the matched pattern for PSUBSCRIBE.

redis://host:port?command=BLPOP&key=jobs&timeout=5 blocks on a list key and submits one Exchange per popped item. The CamelRedis.Key header carries the list key. The timeout parameter is the block duration in seconds. Use BLPOP for left pop and BRPOP for right pop.

The Consumer's mode comes from the URI command. SUBSCRIBE and PSUBSCRIBE use Pub/Sub mode. BLPOP and BRPOP use queue mode. A command that fits neither returns an error at consumer creation. The component does not silently fall back to BLPOP (REDIS-003).

Security

The component supports TLS through the rediss:// URI scheme or the ssl=true parameter. The two are equivalent. TLS auto-enables for non-loopback hosts when the redis crate is compiled with a TLS feature (tls-rustls-webpki-roots, tls-rustls-native-certs, or tls-native-tls). The component logs a tracing::warn! when auto-enabling TLS. A missing feature gate stops startup with the required cargo add command.

The component redacts passwords in Debug output. Passwords with special characters (@, :, /) are percent-encoded in the connection URL. The safe_endpoint() helper returns a credential-free identifier for tracing.

Connection handling

The Producer holds a single multiplexed connection per Endpoint. The Consumer holds one connection for Pub/Sub mode and one for queue mode. Each connection has a 10-second timeout by default. Transient transport errors trigger reconnect with the configured NetworkRetryPolicy. The route ErrorHandler owns the operational signal for non-transient errors.

The component registers an async health check that sends a PING command. The probe is healthy when Redis responds with PONG and degraded when PING fails or times out.

Error handling

The Consumer logs at error! for channel-closed conditions on Pub/Sub and BLPOP send paths and for retry-exhaustion. Each site reports a typed metric before the log line. The Producer logs send failures at warn!. Per-message non-transient Redis errors log at error! with a typed metric. The route handler owns the operational signal for transient producer errors.

Reference: Redis crate CONTEXT. Example source: examples/redis-example.

Database

The database components connect routes to SQL and SurrealDB. Two crates cover the two database families. The SQL component uses sqlx for PostgreSQL, MySQL, and SQLite. The SurrealDB component uses the SurrealDB SDK for document, graph, vector, and live-query operations.

SQL

The SQL component executes queries against relational databases through sqlx. It covers both directions. A Consumer polls query results. A Producer executes statements.

URI

sql:<query>[?outputType=<mode>&dataSource=<name>&allowDynamicQuery=<bool>&bridgeErrorHandler=<bool>]

Consumer

sql:SELECT * FROM users?outputType=SelectList polls the query result. The Consumer submits one Exchange per poll cycle. The body carries the result set as Body::Json.

outputType=StreamList exposes rows as a lazy NDJSON stream. Combined with the streaming split EIP, this gives at-least-once row processing. The Consumer fetches rows from the database as the stream drains. It does not commit the batch until the whole split pipeline completes. If the pipeline crashes mid-stream, the batch re-delivers on the next poll. When a streaming split sub-pipeline uses sql: as a producer, set the pool max_connections to at least 2 to avoid a deadlock.

Producer

sql:UPDATE users SET name = :#name WHERE id = :#id executes a statement with named parameters. The Producer supports positional # and named :#<token> placeholders. Named parameters resolve from the Exchange body, headers, and properties through ExchangeLookupPath.

The allowDynamicQuery parameter defaults to false. In this mode, the Producer ignores the CamelSql.Query header and the Exchange body. It uses only the query from its Endpoint configuration. This default enforces the exchange-data trust boundary (ADR-0032). Set allowDynamicQuery=true to let the route supply the query. The opt-in makes the route responsible for validating the source.

Placeholder syntax

The SQL placeholder parser accepts these forms:

• Positional #. Body must be a JSON array. Bindings come from index.
• Named :#<token>. Resolves from the body JSON tree, headers, or properties.
• IN-clause :#in:<token>. Value must resolve to a JSON array.
• Expression :#${<expr>}. Escape hatch that uses ExchangeLookupPath.
• PostgreSQL ::cast. :#id::text resolves the placeholder and leaves ::text as the SQL cast.

The full contract surface with accepted, rejected, and forbidden patterns lives in the SQL crate CONTEXT.

SurrealDB

The SurrealDB component connects routes to SurrealDB for document, graph, vector, function, and live-query operations. It uses the SurrealDB SDK directly.

URI

surrealdb:<operation>?[dataSource=<name>&...]

The operation is the path segment in the URI. The 12 supported operations are:

Consumer

surrealdb:live?dataSource=my-db subscribes to live query notifications. The Consumer submits one Exchange per live event. Live queries require a WebSocket connection (ws or wss). HTTP connections are rejected for live operations.

Producer

surrealdb:query?dataSource=my-db executes SurrealQL. The query comes from the CamelSurrealDbQuery header, the Exchange body, or the Endpoint configuration. Results materialize as Body::Json. The output=stream option is rejected.

Security

Credentials belong in the named datasource, not in the Endpoint URI. The redact_db_url function removes user information before a URL enters a log or error message. Identifier validation rejects whitespace, quotes, semicolons, and backslashes in interpolated table, edge, and vector-field names.

The raw-query trust boundary is an open hardening gap. The query operation accepts SurrealQL from the CamelSurrealDbQuery header or the body before it falls back to Endpoint configuration. Route authors must filter these sources before the Producer. The SurrealDB crate CONTEXT documents the gap.

Datasource configuration

Both components use named datasources from Camel.toml:

[components.datasources.my-db]
db_url = "postgres://user:pass@localhost/mydb"

The SQL component also supports inline connection parameters in the URI. The SurrealDB component requires credentials in the named datasource.

Reference: SQL crate CONTEXT, SurrealDB crate CONTEXT.

SurrealDB

The SurrealDB component connects routes to a SurrealDB multi-model database. One scheme covers twelve operations: document CRUD, graph edges, vector search, function calls, and live-query change data capture. Both directions share the same surrealdb: scheme. The Consumer is push-based and limited to live; the Producer handles the other eleven operations.

A producer route writes a user and selects the table:

use camel_builder::RouteBuilder;

let create = RouteBuilder::from("timer:tick?period=2000&repeatCount=1")
    .set_body(serde_json::json!({"name": "Alice", "age": 30}))
    .to("surrealdb:create?datasource=demo&table=users")
    .build()?;

let list = RouteBuilder::from("timer:tick?period=2000&repeatCount=1&delay=500")
    .to("surrealdb:select?datasource=demo&table=users")
    .build()?;

routes:
  - id: create-user
    from: "timer:tick?period=2000&repeatCount=1"
    steps:
      - set_body: '{"name":"Alice","age":30}'
      - to: "surrealdb:create?datasource=demo&table=users"

  - id: list-users
    from: "timer:tick?period=2000&repeatCount=1&delay=500"
    steps:
      - to: "surrealdb:select?datasource=demo&table=users"

A live-query route turns table changes into a stream of Exchanges:

let live = RouteBuilder::from("surrealdb:live?datasource=demo&table=events")
    .to("log:info?showBody=true")
    .build()?;

routes:
  - id: events-cdc
    from: "surrealdb:live?datasource=demo&table=events"
    steps:
      - to: "log:info?showBody=true"

URI

surrealdb:<operation>?datasource=<name>[&table=<t>][&id=<r>][&edge=<e>][&from=<r>][&to=<r>][&function=<f>][&top_k=<n>][&metric=<m>][&vector_field=<f>][&query=<surrealql>]

The path segment is the operation. The component rejects the URI if datasource is missing, the operation is unknown, or output=stream is set. Streaming output is not supported. All results materialize as Body::Json(Vec<Value>).

Common parameters

Retry policy parameters (retryEnabled, retryMaxAttempts, retryInitialDelayMs, retryMultiplier, retryMaxDelayMs, retryJitter) match camel-sql by name and default. They configure pool-establishment retry today. ADR-0013 owns the retry semantics; producer operations do not retry to avoid duplicating non-idempotent writes.

Producer

The Producer executes the eleven non-live operations. The body shape the Producer requires depends on the operation. create, update, upsert, patch, relate, vector, and run require a Body::Json body. query, select, search, and delete accept input through URI parameters or headers, so an empty body is valid.

select and query also expose a PollingConsumer. They integrate with the Poll Enrich EIP for on-demand reads. Other operations do not poll.

Result bodies land as Body::Json. The Producer sets CamelSurrealDbRecordId on writes when the id can be derived from the URI or the response. The id is the edge record for relate, not the source node.

The query operation reads SurrealQL from the URI, then the CamelSurrealDbQuery header, then the Exchange body. The CamelSurrealDbParams header binds $name placeholders. ADR-0032 classifies Exchange data as untrusted. The component has no allow_dynamic_query switch. The route is responsible for sanitizing SurrealQL from external sources before it reaches the Producer.

Consumer

surrealdb:live?datasource=<name>&table=<t> subscribes to a SurrealDB live query. SurrealDB pushes one notification per CREATE, UPDATE, or DELETE on the table. The Consumer submits one Exchange per notification. The body holds the affected record. The Consumer sets CamelSurrealDbAction to the action type and CamelSurrealDbTable to the table name.

Live queries require a WebSocket transport (ws:// or wss://). The component rejects http:// and https:// datasources for the live operation at endpoint creation.

The Consumer is event-driven. It does not implement PollingConsumer. The live operation exposes no receive() entry point.

Datasource

The datasource URI parameter names a Camel.toml entry. Credentials belong in the datasource, not the URI. The component redacts user information from db_url before any URL enters a log line or error message.

[default.datasources.demo]
db_url = "ws://localhost:8000"
provider = "surrealdb"

[default.datasources.demo.extra]
namespace = "test"
database = "test"
username = "root"
password = "root"

The extra map carries namespace, database, username, and password. The transport schemes are ws://, wss://, http://, and https://. The live operation requires ws or wss. See the Database page for the shared datasource contract.

Headers

Security

Identifier validation runs on table, edge, and vector_field. The validator accepts ASCII letters, digits, and _. The first character must be a letter or _. Whitespace, quotes, semicolons, and backslashes are rejected. Record keys (id, from, to) go through SDK binding and accept the broader charset SurrealDB requires for numeric, UUID, and string keys.

from and to for relate must be full RecordIds in table:key form. Bare keys are rejected at endpoint creation. This prevents the edge from silently targeting the wrong record when a route author writes from=1 instead of from=user:1.

The raw-query trust boundary is the documented hardening gap for this component. The query operation accepts SurrealQL from the body and the CamelSurrealDbQuery header. The route must filter these sources before the Producer. The component exposes no default-deny switch equivalent to camel-sql's allowDynamicQuery=false.

The component follows the ADR-0012 log-level convention. It never emits error! in production code. Recoverable and handler-owned conditions use warn!. Errors that terminate an operation return CamelError or SurrealDbError. The route error handler or supervision owns the final signal. ADR-0020 classifies the SurrealDB SDK as a stable database-driver class. The crate does not wrap the SDK in a project-owned adapter trait.

Reference: SurrealDB crate CONTEXT. Example source: examples/surrealdb-example.

OpenSearch

The OpenSearch component runs document, index, and search operations against an OpenSearch cluster. The OpenSearchProducer is a Tower Service<Exchange> that owns a shared client and a 128-permit semaphore. The MULTISEARCH operation is reserved but not yet implemented. The component is producer-only. The Endpoint rejects create_consumer() with EndpointCreationFailed.

Index a document

The Producer sends the Exchange body as the document. The optional CamelOpenSearch.Id header sets the document ID. Without it, OpenSearch assigns a server-generated ID.

use camel_api::Exchange;
use camel_builder::RouteBuilder;

let route = RouteBuilder::from("direct:index-user")
    .set_header("CamelOpenSearch.Id", serde_json::json!("user-42"))
    .set_body(serde_json::json!({
        "name": "Alice",
        "age": 30
    }))
    .to("opensearch://localhost:9200/users?operation=INDEX")
    .build()?;

routes:
  - id: index-user
    from: "direct:index-user"
    steps:
      - set_header:
          CamelOpenSearch.Id: "user-42"
      - set_body:
          name: "Alice"
          age: 30
      - to: "opensearch://localhost:9200/users?operation=INDEX"

Search documents

The Producer sends the body as the search query. The size and from URI params fill in pagination when the body does not set them.

use camel_builder::RouteBuilder;

let route = RouteBuilder::from("direct:search-users")
    .set_body(serde_json::json!({
        "query": {
            "match": { "name": "Alice" }
        }
    }))
    .to("opensearch://localhost:9200/users?operation=SEARCH&size=25&from=0")
    .build()?;

routes:
  - id: search-users
    from: "direct:search-users"
    steps:
      - set_body:
          query:
            match:
              name: "Alice"
      - to: "opensearch://localhost:9200/users?operation=SEARCH&size=25&from=0"

Bulk operations

The body is a JSON array of action-and-document pairs. The Producer serializes each element to a line and rejects the call when the total payload exceeds max_bulk_bytes.

use camel_builder::RouteBuilder;

let route = RouteBuilder::from("direct:bulk")
    .set_body(serde_json::json!([
        { "index": { "_id": "1" } },
        { "name": "Alice", "age": 30 },
        { "index": { "_id": "2" } },
        { "name": "Bob", "age": 25 }
    ]))
    .to("opensearch://localhost:9200/users?operation=BULK&max_bulk_bytes=1048576")
    .build()?;

routes:
  - id: bulk
    from: "direct:bulk"
    steps:
      - set_body:
          - index:
              _id: "1"
          - name: "Alice"
            age: 30
          - index:
              _id: "2"
          - name: "Bob"
            age: 25
      - to: "opensearch://localhost:9200/users?operation=BULK&max_bulk_bytes=1048576"

URI

opensearch://<host>:<port>/<index>?operation=<op>[&option=value...]
opensearchs://<host>:<port>/<index>?operation=<op>[&option=value...]  (TLS)

Operations

The Producer selects the operation from the CamelOpenSearch.Operation header. The header overrides the URI operation param. Invalid header values fall back to the URI value. URI parsing rejects unknown operation names at config time.

Headers

The Producer reads the index name from operator endpoint configuration. URI parsing validates the index name against OpenSearch naming rules and fails closed on null bytes, traversal segments, and invalid characters. The CamelOpenSearch.Id header is untrusted Exchange data. Current typed request builders pass it to _doc/{id} paths without component-side validation. This gap is tracked as audit finding I1.

TLS

Use the opensearchs scheme to enable HTTPS. The base URL builder selects https:// from the scheme. Basic auth credentials apply to both schemes.

routes:
  - id: secure-search
    from: "direct:search-secure"
    steps:
      - to: "opensearchs://opensearch.example.com:9200/users?operation=SEARCH&username=admin&password=secret"

Global configuration

Set defaults in Camel.toml under the opensearch key. URI params always override global values. Unset URI fields fall back to the global config after URI parsing.

[default.components.opensearch]
host = "opensearch.internal"
port = 9200
username = "app-user"
password = "${OPENSEARCH_PASSWORD}"
default_operation = "SEARCH"
index_name = "events"
timeout_ms = 30000

The OpenSearchConfig::Debug implementation redacts the password to <redacted>. The same redaction applies to OpenSearchEndpointConfig::Debug.

Concurrency and retry

The Producer holds a Semaphore that caps in-flight calls at 128. poll_ready() reserves a permit before the call. A closed semaphore maps to CamelError::ConsumerStopping (ADR-0019, ADR-0024). The shared client initializes lazily on the first call. Arc<Mutex<Option<OpenSearch>>> serializes initialization, then each call clones and reuses the client.

Retry uses NetworkRetryPolicy. The retry loop classifies failures as transient or permanent:

• Transient: HTTP 5xx, network failures, timeouts. Retried up to max_attempts
• Permanent: HTTP 4xx, missing headers, parse failures. Surfaced immediately

All operations except MULTISEARCH and UNKNOWN use typed request builders from the opensearch crate. The component never interpolates Exchange data into query text. Bodies become structured JSON values that serde supplies to the request builder.

Error handling

Operation failures log at warn! with the index name and error. Client initialization failures log at error! because the producer cannot function without a client. ADR-0012 classifies initialization failures as system-broken. The Producer surfaces transient retry exhaustion and permanent failures as CamelError::ProcessorError. The route ErrorHandler owns the operational signal.

Reference: camel-opensearch CONTEXT. Example source: not yet published.

LLM

The LLM component sends chat, embedding, and tool-calling requests to language model providers. It streams tokens or materializes full responses. Three providers ship: OpenAI, Ollama, and Mock. The component is producer-only. It has no Consumer in the current release.

The llm-example registers a Mock provider and runs three routes. The materialized route below sends a prompt, collects the response, and logs it with headers:

let route = RouteBuilder::from("timer:tick?period=3000")
    .route_id("llm-materialized")
    .set_body("Explain Tower middleware in one sentence.")
    .to("llm:chat?provider=openai-prod&model=gpt-4o&stream=false")
    .to("log:info?showBody=true&showHeaders=true")
    .build()?;
ctx.add_route_definition(route).await?;

routes:
  - id: llm-materialized
    from: "timer:tick?period=3000"
    steps:
      - set_body: "Explain Tower middleware in one sentence."
      - to: "llm:chat?provider=openai-prod&model=gpt-4o&stream=false"
      - to: "log:info?showBody=true&showHeaders=true"

URI

llm:<operation>?provider=<name>[&model=<model>][&temperature=<n>][&max_tokens=<n>][&stream=<bool>][&system_prompt=<text>][&timeout_secs=<n>]

Operations

llm:chat runs a chat completion. llm:embed generates a vector embedding. The operation picks which LlmProvider method the producer calls.

Producer

llm:chat?provider=openai-prod&model=gpt-4o sends the Exchange body as a user message. The stream parameter selects streaming or materialized mode.

Streaming mode is the default. The producer returns Body::Stream(StreamBody). Each ChatEvent::Delta carries a token or text fragment. Usage metadata goes to tracing::info!. Usage metadata does not go to Exchange headers. The header CamelLlmUsageAvailable is false at Exchange return time.

Materialized mode (stream=false) collects every ChatEvent into a single Body::Text. The producer writes token counts to CamelLlmTokensIn and CamelLlmTokensOut. It writes the finish reason to CamelLlmFinishReason. The CamelLlmUsageAvailable header is true.

Tool calls flow through CamelLlmTools (input) and CamelLlmToolCalls (output). The component emits tool-call intent through ChatEvent::ToolCall. The component never executes tools. The route owns dispatch.

Cost and cache

Cost observability tracks spending per request. A config-driven PricingTable maps input and output tokens to USD rates. The producer computes cost from final usage. It writes cost to CamelLlmEstimatedCostUsd in materialized mode. It logs cost at info! in both modes. Missing pricing means no cost header and no failure.

The response cache deduplicates materialized requests. It sits at the producer level, not the provider level. A single-flight mechanism built on dashmap and tokio::sync::watch collapses concurrent lookups. Entries expire by TTL and evict by LRU. The cache stores usage, not cost. The cache lookup runs before the semaphore, retry, and timeout layers.

Providers

The Mock provider ships in the default build. OpenAI and Ollama require cargo features (--features openai and --features ollama). The SiumaiProvider adapter isolates the siumai SDK. If siumai breaks, only provider/siumai_adapter.rs changes (ADR-0020).

Retry and timeout

The manual retry loop honors provider retry_after over exponential backoff (ADR-0021). The loop only runs in materialized mode. Once content starts streaming, the loop stops.

The streaming timeout_secs covers each stream.next() call. A deadline lapse yields a Timeout error. The materialized total deadline covers the whole request.

Error handling

LlmError maps to three CamelError variants: Unauthenticated, Io, and ProcessorError. The mapping is lossy. Downstream handlers that need typed LLM errors should use CamelError::ProcessorErrorWithSource. The component has no Consumer. It rejects create_consumer with CamelError::InvalidUri.

The producer logs provider errors at warn!. The route ErrorHandler owns the error! level. Stream finish and tool dispatch log at info!. Cache hit and miss log at debug!.

Reference: LLM crate CONTEXT, ADR-0020, ADR-0021. Example source: examples/llm-example.

WASM

The WASM component runs operator-installed WebAssembly plugins in a Wasmtime sandbox. One crate covers the plugin, bean, authorization-policy, and source worlds. Each world receives a different capability grant. The sandbox limits damage from guest defects. The sandbox is not a security boundary for intentionally malicious plugins.

The wasm-example wires a timer-driven producer that calls an echo plugin, then logs the result:

    let route = RouteBuilder::from("timer:tick?period=1000&repeatCount=3")
        .route_id("wasm-example")
        .to("wasm:echo.wasm?timeout=5&max-memory=10485760")
        .to("log:info")
        .build()?;

routes:
  - id: wasm-example
    from: "timer:tick?period=1000&repeatCount=3"
    steps:
      - to: "wasm:echo.wasm?timeout=5&max-memory=10485760"
      - to: "log:info"

URI

wasm:<path/to/module.wasm>[?<param>=<value>...]

The path must be relative. Absolute paths and .. segments are rejected. The path resolves against the base directory passed to WasmComponent::new.

Zero or invalid values fall back to the runtime default. The component does not hide defaults. ADR-0011 names the rule.

Worlds

The component ships four WIT worlds. Each world grants a different capability set.

The authorization-policy world also backs the SecurityPolicy host type. The capability grant stays denied() for both roles.

The source world grants no Camel host functions. The host binds the TCP listener and hands the guest an http-listener resource handle. The guest then drives accept-http and submit-exchange on its own loop under Store::run_concurrent. ADR-0031 defines the source lifecycle.

Capability model

ADR-0050 defines the target capability posture for the WASI surface. The current linker registers the full WASI 0.2 surface. The runtime context denies filesystem preopens, environment variables, sockets, and IP-name lookup. Clocks and random remain usable. Processor, bean, and policy guests inherit host stderr. Source guests do not.

ADR-0050 selects per-world selective WASI registration as the target. The migration is in progress. The runtime grants today match the target grants. The linker surface is broader until the migration lands. A Wasmtime upgrade or a WasiCtxBuilder change must not widen capabilities by accident (ADR-0050).

The Camel host function grants depend on the world. The WasmCapabilities struct carries two fields.

Authorization and security policy guests use WasmCapabilities::denied(). Processor and bean guests use WasmCapabilities::from_scheme_list(schemes). from_scheme_list sets host_kv to true. Policy guests do not get host storage.

The allow-call URI parameter and the allow-call-schemes Camel.toml field both flow into call_schemes. An empty list denies every scheme. The component fails closed (ADR-0033).

Configuration via Camel.toml

Processor plugins read limits from the URI query string. Bean, authorization-policy, and security-policy plugins read limits from a [limits] block in Camel.toml. The block type is WasmLimitsConfig. ADR-0014 unifies these knobs across plugin kinds.

[default.beans.my-bean]
plugin = "my-bean"

[default.beans.my-bean.limits]
timeout-secs = 600
max-memory = 4294967296
max-concurrent-calls = 1

[security.permissions.providers.my-policy]
provider = "wasm"
path = "plugins/authz.wasm"

[security.permissions.providers.my-policy.limits]
timeout-secs = 5
max-memory = 10485760

All fields are optional. None means use the runtime default. WasmConfig::from_limits is the single source of truth for defaults. No silent fallback lie exists elsewhere (ADR-0011).

Security

The component trusts the plugin the operator installs. The sandbox limits damage from guest defects. The sandbox is not a security boundary for intentionally malicious plugins. ADR-0032 classifies Exchange data as untrusted. The sandbox contains the guest while it processes that data.

Path validation rejects absolute paths. It rejects .. segments. The canonical path must start with the base directory. The DepthGuard prevents camel_call from re-entering the same Store. The guard releases the recursion count on return, error, or cancellation.

init_config sends trusted operator configuration to the guest. Operators must not place secrets there unless the guest needs them. Debug output for init_config or StateStore must not expose secret values.

A known gap remains: set_property and host-side StateStore allocations do not have independent size limits. Wasmtime store limits do not account for these host allocations. The finding is F-camel-component-wasm-I4.

Error handling

WASM transform execution inside the pipeline logs at warn!. The route ErrorHandler owns the error. The producer downgrades the level to warn! with the // log-policy: handler-owned marker. ADR-0012 classifies this as category (a).

After a Timeout, Trap, or OutOfMemory, the plugin runtime resets on the next call. The route does not need manual intervention.

Reference: WASM crate CONTEXT — ADR-0050: WASM sandbox capability posture. Example source: examples/wasm-example.

Cron

The cron component fires Exchanges on a Unix 5-field cron schedule. Use it for source routes that run at calendar times, not fixed intervals. The component mirrors the timer structure (Component to Endpoint to Consumer) but delegates scheduling to a pluggable CronService.

The cron-example fires a route every minute:

    let route = RouteBuilder::from("cron:tick?schedule=*+*+*+*+*")
        .route_id("cron-demo")
        .set_body("cron-fired")
        .set_header("source", Value::String("cron".into()))
        .to("log:cron-result?level=info&showBody=true&showHeaders=true")
        .build()?;

routes:
  - id: cron-demo
    from: "cron:tick?schedule=*+*+*+*+*"
    steps:
      - set_body: "cron-fired"
      - set_header:
          key: source
          value: "cron"
      - to: "log:cron-result?level=info&showBody=true&showHeaders=true"

URI

cron:<name>?schedule=<5-field-expr>[&timeZone=<IANA>&includeMetadata=true]

Schedule format

The schedule is a standard Unix 5-field cron expression:

minute hour day-of-month month day-of-week

The URI uses + as the space separator (Apache Camel convention). The expression *+*+*+*+* fires every minute. Each field supports the standard operators: *, ranges (1-5), lists (1,3,5), and step values (*/2).

Consumer

cron:tick?schedule=*+*+*+*+* fires an Exchange on the schedule. The Consumer is a pure source. It reads from no external system. The first fire happens at the next matching schedule time, not at startup.

The Consumer delegates scheduling to Arc<dyn CronService>. The default implementation is TokioCronService. The service picks the fire time. The CronConsumer callback submits the Exchange.

Misfire behavior

A missed fire is not replayed. When the process is down at a scheduled fire time, the Consumer recomputes the next fire from the current time. This behavior is misfire-skip. Automatic catch-up of batch jobs is dangerous, so the component skips missed fires instead.

Metadata

Set includeMetadata=true to attach CronFire metadata to each Exchange:

• scheduled_at. The time the fire was scheduled.
• fired_at. The actual time the Consumer submitted the Exchange. This differs from scheduled_at under load.
• counter. The number of fires since the Consumer started.

Error propagation

The CronCallback is an async, fallible closure. On Err, the error propagates to Route supervision (ADR-0007). The Route records the failure and, when a supervision policy is configured, restarts the Route.

Reference: cron crate CONTEXT. Example source: examples/cron-example.

Direct

The Direct component routes an Exchange between two routes in the same CamelContext over an in-memory channel. The Producer blocks until the Consumer's Pipeline finishes. No serialization, no network. The transformed Exchange returns to the caller.

The multi-route-direct example wires a timer-driven producer and a transform consumer:

use camel_api::body::Body;
use camel_api::Value;
use camel_builder::RouteBuilder;
use camel_component_direct::DirectComponent;
use camel_component_log::LogComponent;
use camel_component_timer::TimerComponent;
use camel_core::context::CamelContext;

#[tokio::main]
async fn main() -> Result<(), camel_api::CamelError> {
    let mut ctx = CamelContext::builder().build().await.unwrap();
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());
    ctx.register_component(DirectComponent::new());

    // Route A: timer -> direct:pipeline
    let route_a = RouteBuilder::from("timer:tick?period=1000")
        .route_id("route-a")
        .set_header("source", Value::String("timer".into()))
        .to("direct:pipeline")
        .build()?;

    // Route B: direct:pipeline -> uppercase -> log
    let route_b = RouteBuilder::from("direct:pipeline")
        .route_id("route-b")
        .map_body(|body: Body| {
            if let Some(text) = body.as_text() {
                Body::Text(text.to_uppercase())
            } else {
                body
            }
        })
        .to("log:output?showBody=true")
        .build()?;

    ctx.add_route_definition(route_a).await?;
    ctx.add_route_definition(route_b).await?;
    ctx.start().await?;
    Ok(())
}

routes:
  - id: route-a
    from: "timer:tick?period=1000"
    steps:
      - set_header:
          key: "source"
          value: "timer"
      - to: "direct:pipeline"
  - id: route-b
    from: "direct:pipeline"
    steps:
      - to: "log:output?showBody=true"

URI

direct:<name>[?timeout_ms=30000][&failIfNoConsumers=true]

The endpoint name must not be empty and must not contain whitespace. The Component rejects both at Endpoint creation.

Consumer

from: "direct:name" registers a DirectConsumer in the shared registry. The Consumer starts a background loop that receives Exchanges from the in-memory channel. It submits each Exchange to the Route's Pipeline through send_and_wait. The reply carries the transformed Exchange or the failure.

One Consumer per name. A second Consumer on the same name returns CamelError::EndpointCreationFailed because the registry already holds an open channel. Routes that want many workers must use a different name, or pick a Component that supports fanout (SEDA with multipleConsumers=true).

The Consumer removes its registry entry on stop() or when the cancellation token fires. The next Producer call fails with EndpointCreationFailed until a new route registers.

Producer

to: "direct:name" builds a DirectProducer. The Producer holds one in-flight call at a time through a bounded semaphore. poll_ready checks the registry and acquires a permit. call hands the Exchange to the Consumer's channel and awaits the reply.

failIfNoConsumers=true (default) rejects the call when no Consumer is registered. Set it to false to let the Producer race against late registration. The Producer still waits for the Consumer to receive the Exchange, so false does not give a fire-and-forget guarantee. Use SEDA for that.

The default timeout_ms is 30 000. A timeout returns CamelError::ProcessorError with a timed out message. The error propagates through the route's error handler.

Request-Reply

Direct is the natural request-reply Component for in-process calls. The Producer blocks until the Consumer's Pipeline finishes. Steps that follow to: "direct:name" see the transformed Exchange. Steps that set the body inside the Consumer's Route reach the caller. A timer that sends to a Direct endpoint and logs the result is a synchronous in-process function call.

Error handling

The DirectConsumer reports unhandled pipeline failures through the b-prime:direct:send-and-wait metric and logs at error! (ADR-0012 category b'). Producer send failures (no Consumer, channel closed, reply dropped) are category (a) handler-owned. The Producer logs them at warn!. The route's error handler owns the operational signal.

Reference: Direct crate CONTEXT. Example source: examples/multi-route-direct.

SEDA

The SEDA (Staged Event-Driven Architecture) component stages exchanges in memory between routes that share one CamelContext. A producer sends to seda:name and returns. A consumer on the same name pulls from a bounded queue and processes asynchronously.

seda: is the asynchronous counterpart to direct:. Reach for it to decouple route lifetimes, smooth traffic bursts, or fan out to multiple subscribers.

The seda-demo wires a timer-driven producer against an asynchronous consumer that uppercases the body:

use camel_api::body::Body;
use camel_builder::RouteBuilder;
use camel_component_log::LogComponent;
use camel_component_seda::SedaComponent;
use camel_component_timer::TimerComponent;
use camel_core::context::CamelContext;

let mut ctx = CamelContext::builder().build().await.unwrap(); // allow-unwrap
ctx.register_component(TimerComponent::new());
ctx.register_component(LogComponent::new());
ctx.register_component(SedaComponent::new());

let route_a = RouteBuilder::from("timer:tick?period=1000&repeatCount=5")
    .route_id("producer-route")
    .to("seda:processing")
    .build()?;

let route_b = RouteBuilder::from("seda:processing?concurrentConsumers=2")
    .route_id("consumer-route")
    .map_body(|body: Body| {
        if let Some(text) = body.as_text() {
            Body::Text(text.to_uppercase())
        } else {
            body
        }
    })
    .to("log:output?showBody=true&showHeaders=true")
    .build()?;

ctx.add_route_definition(route_a).await?;
ctx.add_route_definition(route_b).await?;
ctx.start().await?;

routes:
  - id: producer-route
    from: "timer:tick?period=1000&repeatCount=5"
    steps:
      - to: "seda:processing"
  - id: consumer-route
    from: "seda:processing?concurrentConsumers=2"
    steps:
      - to: "log:output?showBody=true&showHeaders=true"

URI

seda:<name>[?size=<n>][&concurrentConsumers=<n>][&multipleConsumers=<bool>][&blockWhenFull=<bool>][&discardIfNoConsumers=<bool>][&timeout=<ms>][&waitForTaskToComplete=<mode>][&exchangePattern=<pattern>]

Endpoints that share a name must agree on size, multipleConsumers, exchangePattern, and concurrentConsumers. The component rejects mismatched shared options with an EndpointCreationFailed error.

Consumer

seda:processing?concurrentConsumers=2 registers a consumer that pulls from the endpoint's bounded queue. The Runtime starts one queue forwarder task per consumer. That forwarder awaits send_and_wait for InOut and waitForTaskToComplete=Always exchanges, so those exchanges remain serial even when concurrentConsumers is greater than 1. InOnly exchanges without a reply channel do not block the forwarder.

concurrentConsumers is reported to the Runtime through ConcurrencyModel::Concurrent. Finding I1 and bd issue rc-exa2 track the limitation that blocks true concurrent InOut processing.

The consumer transfers the primary forwarder handle to the Runtime through background_task_handle(). On shutdown, the Runtime aborts that handle, then calls stop(). stop() cancels the private token, aborts retained forwarders, and clears the active consumer registration.

Producer

seda:processing creates a producer that enqueues the exchange. The producer returns immediately for fire-and-forget patterns. Behavior depends on waitForTaskToComplete:

• Never returns after enqueue. No reply channel is attached.
• IfReplyExpected waits only when exchangePattern=InOut.
• Always waits regardless of the exchange pattern.

When the producer waits, the forwarder on the consumer side uses send_and_wait to route the pipeline result back through a oneshot channel. A timeout returns EndpointCreationFailed. A closed channel returns ChannelClosed.

A full queue returns EndpointCreationFailed with the queue name and size. blockWhenFull=true makes the producer wait up to timeout for capacity. The route ErrorHandler owns both signals per ADR-0019.

Modes

SedaMode::Single owns one queue and permits one active consumer. A second registration on the same name returns an EndpointCreationFailed error.

SedaMode::Fanout, enabled by multipleConsumers=true, owns one queue per subscriber. A fanout producer reserves capacity for all subscribers before it sends, so delivery is all-or-nothing for the active subscriber set. Fanout rejects reply-waiting modes because one request has no single valid reply. The combination multipleConsumers=true with waitForTaskToComplete=Never is the only legal configuration.

SEDA vs Direct

Reach for direct: when two routes must share a call stack and ordering is strict. Reach for seda: when you need to decouple producer and consumer lifetimes, smooth bursts, or broadcast to multiple subscribers.

Error handling

A producer that targets an endpoint with no active consumer returns EndpointCreationFailed with the message SEDA endpoint '<name>' has no active consumers. Set discardIfNoConsumers=true to drop silently in this case.

A full queue with blockWhenFull=false returns EndpointCreationFailed with the queue name and configured size. Set blockWhenFull=true to wait up to timeout.

Route stop does not drain an in-flight reply. An interrupted InOut producer can receive CamelError::ChannelClosed. This is the current best-effort contract for in-memory staging. It is not an ADR-0004 hot-reload pipeline swap.

The public ExchangePattern and WaitForTaskToComplete enums are closed URI option sets. They stay exhaustive. ADR-0049 does not bind this component crate.

Reference: SEDA crate CONTEXT. Example source: examples/seda-demo.

ControlBus

The ControlBus component sends Route lifecycle commands through the RuntimeBus. It is a Producer-only Endpoint. It does not consume. It exposes no network API.

The controlbus-example uses a timer to suspend and resume a target route:

    let suspend_route = RouteBuilder::from("timer:suspend?delay=5000&repeatCount=1")
        .route_id("suspend-controller")
        .process(|exchange| async move {
            println!("[CONTROL] Suspending target-route...");
            Ok(exchange)
        })
        .to("controlbus:route?routeId=target-route&action=suspend&authorizedRoutes=target-route")
        .to("log:control?showBody=true")
        .build()?;

routes:
  - id: target-route
    from: "timer:target?period=500"
    steps:
      - to: "log:target?showBody=true"
  - id: suspend-controller
    from: "timer:suspend?delay=5000&repeatCount=1"
    steps:
      - to: "controlbus:route?routeId=target-route&action=suspend&authorizedRoutes=target-route"
      - to: "log:control?showBody=true"

URI

controlbus:route?routeId=<id>&action=<action>&authorizedRoutes=<csv>

Actions

restart performs an atomic Pipeline swap without drain semantics (ADR-0004). Suspend and resume support varies by component. status returns only the lifecycle string, not Route statistics.

Authorization

The Producer enforces three gates on every call (ADR-0034):

1. The URI declares the target routeId.
2. authorizedRoutes exists and contains that target.
3. The target differs from the calling Route ID.

The CamelRouteId Exchange header cannot select or override the target. Exchange data is untrusted (ADR-0032). Only operator configuration drives the control plane. Authorization failures return CamelError::Unauthorized.

Errors

The component declares no public enums. camel-api provides RouteAction, RuntimeCommand, and CamelError. Future variants use fallback match arms (ADR-0049).

Reference: ControlBus crate CONTEXT. Example source: examples/controlbus.

Master

The Master component runs a delegate Consumer only on the node that holds a leadership lock. Other nodes join the same lock name and stand by until they win. ADR-0035 establishes the leader-epoch fencing token that the bridge stamps on every emitted envelope.

    let route_1 = RouteBuilder::from("master:mylock:timer:tick?period=1000")
        .route_id("master-route")
        .to("log:info")
        .build()?;

    let route_2 = RouteBuilder::from("timer:status?period=5000")
        .route_id("status-route")
        .to("controlbus:route?routeId=master-route&action=status")
        .to("log:info")
        .build()?;

    ctx.add_route_definition(route_1).await?;
    ctx.add_route_definition(route_2).await?;

The example drives a timer:tick source through a master:mylock lock. Only the elected leader consumes ticks. A second route uses ControlBus to poll the leader's status every five seconds.

routes:
  - id: master-route
    from: "master:mylock:timer:tick?period=1000"
    steps:
      - log: "DEBUG: tick"

URI

master:<lockname>:<delegate-uri>

The Master has no URI query parameters of its own. Query parameters belong to the delegate Component.

Consumer

master:<lockname>:<delegate-uri> gates a delegate Consumer on leadership. When the route starts, the node joins leader election under the lock name. The delegate Consumer does not start until the node wins. On leadership loss, the delegate stops and drains within drain_timeout_ms. If the node wins again, the delegate restarts with the same delegate URI.

Each ExchangeEnvelope the delegate emits carries an x-camel-leader-epoch Exchange property. The Master stamps the property with a monotonic fencing token at bridge start. A stale bridge retains the epoch from its own spawn, not the live epoch. Downstream sinks that need split-brain safety compare the property against the current leader's epoch and reject older envelopes.

Producer

to("master:mylock:http://api.example.com") passes through to the delegate Producer without leader gating. The Master's job is to gate Consumers, not Producers. A write that needs exclusive access should serialize through a queue that the leader Consumer drains.

Leadership backends

The leader election backend comes from the configured PlatformService. The default NoopPlatformService always elects the local node, so a single-node deployment works without external infrastructure. A Kubernetes deployment uses KubernetesPlatformService and Lease objects for distributed election across pods.

Every acquired term increments the leader epoch. The bridge stamps the new epoch on each envelope. A node that loses leadership stops the delegate within drain_timeout_ms and steps down. The route stays alive; only delegate intake pauses until the node wins again.

Configuration

The Master reads from [components.master] in Camel.toml:

When both reconnect and delegate_retry_max_attempts are set, the explicit reconnect value wins. The delegate_retry_max_attempts field stays for backward compatibility with earlier configs.

Reference: Master component CONTEXT. Example source: examples/master-leader and examples/master-leader-yaml.

Template

The Template component renders MiniJinja templates loaded from the filesystem against the body, headers, and properties of each inbound exchange. It is producer-only — you place it on the to: side of a route to transform the exchange body into rendered output.

The template-basic example wires a timer-driven source through a header set and the template Producer:

    // Route: timer → set body/headers → render template → log result
    let route = RouteBuilder::from("timer:tick?period=2000")
        .route_id("template-demo")
        .set_body("World")
        .set_header("title", Value::String("Template Demo".into()))
        .to(&template_uri)
        .log("Rendered template", LogLevel::Info)
        .build()?;

routes:
  - id: template-demo
    from: "timer:tick?period=2000"
    steps:
      - set_body: "World"
      - set_header:
          title: "Template Demo"
      - to: "template:file:///srv/templates/page.html.tmpl"
      - log: "Rendered template"

URI

template:file:///<absolute-path-to-template>

The URI has two parts. The outer scheme is template. The inner scheme is file, followed by an empty authority and an absolute path. Bare paths (template:/srv/t/page.html) and non-file inner schemes are rejected at Endpoint construction. The path must be absolute and free of .. segments.

The component is zero-override. The URI is operator-configured at route construction. No Exchange header or property can replace the entry, the root, or the template source. The template:file:/// form is the only accepted shape; there is no template:http:// or header-driven loader.

Render contract

The Producer replaces exchange.input.body with the rendered output as Body::Text. Headers and properties are preserved unchanged. On any render failure — strict-undefined variable, output-size overflow, fuel exhaustion, recursion limit, or execution timeout — the body is left byte-identical to the inbound value. The route ErrorHandler then owns the operational signal.

The rendering context exposes three top-level keys: body, headers, and exchangeProperty. Body::Text and Body::Bytes (lossy UTF-8) are accepted. Body::Json exposes its fields as {{ body.field }}. Body::Stream is rejected with a guidance error pointing to stream_cache upstream. Body::Empty renders to an empty string without tripping strict-undefined.

Every template must declare a top-level {% autoescape %} block selecting html, json, or none. The component enforces the explicit declaration at compile time. There is no global default; per-render output context is an operator decision, not a framework assumption.

Limits

The bundle owns two independent limit layers under [components.template]. Both default when absent. A zero value is rejected at startup; a limit cannot be silently disabled.

[components.template.limits]
max-total-source-bytes = 16777216
max-include-count = 64
max-include-depth = 16
max-template-size = 1048572
reload-timeout-ms = 5000

[components.template.render-limits]
max-context-size = 65536
max-output-size = 1048576
fuel = 100000

The two layers are checked at different times. Acquisition limits apply while building the compiled template set. Render limits apply per Exchange on the hot path. Exhausting any limit fails the operation; no limit truncates and reports success.

Lifecycle

Templates are compiled once at route startup (fail-closed). A missing file, a syntax error, or an acquisition-limit overflow prevents the route from starting. The compiled set is cached for zero-filesystem-I/O hot-path rendering. A control-plane ReloadTemplates command re-acquires the dependency closure, recompiles, and atomically swaps the compiled set without disturbing in-flight renders. A failed reload preserves the prior set.

File reads go through openat-relative handles. .. segments, symlinks, absolute path escapes, and include cycles are rejected at acquisition time. The MiniJinja environment registers no functions, filters, or globals — an unknown function call fails at render, not at the host boundary.

When to use it

Use the Template component when the rendered output needs blocks, loops, filters, macros, or context-aware escaping. Use the inline minijinja language under set_body: { language: minijinja, source: ... } for one-shot substitution where the template fits in a route definition. Use a different component when the transformation is XML, JSON bridging, or parameter-bound SQL.

The atomic-write contract and the accepted fileExist values live in the camel-template CONTEXT. The architectural rationale is in ADR-0047. Example source: examples/template-basic.

Validator

The Validator component validates message bodies against XSD, JSON Schema, and YAML Schema files. It runs in routes as a to: Producer. Validation failure returns a CamelError to the route's error handler.

The validator-example wires timer-driven producers against local schemas for each format:

    let route_xsd = RouteBuilder::from("timer:xsd-valid?period=3000&repeatCount=2")
        .route_id("xsd-valid")
        .set_body("<order><id>A1</id><amount>5</amount></order>")
        .log("Route 1: Validating XML order against XSD", LogLevel::Info)
        .validate(&xsd)
        .log("Route 1: XML is valid!", LogLevel::Info)
        .to("log:info?showBody=true")
        .build()?;

routes:
  - id: xsd-valid
    from: "timer:xsd-valid?period=3000&repeatCount=2"
    steps:
      - set_body: "<order><id>A1</id><amount>5</amount></order>"
      - log: "Validating XML order against XSD"
      - to: "validator:schemas/order.xsd"
      - log: "XML is valid"
      - to: "log:info?showBody=true"

URI

validator:<schema-path>[?type=xml|json|yaml&failOnNullBody=true|false&headerName=<name>&failOnNullHeader=true|false&maxPayloadBytes=<n>&schemaCacheMaxEntries=<n>]

The schema type defaults to the file extension (.xsd, .json, .yaml, .yml). Pass type=xml to override the extension when the path is ambiguous. Paths accept percent-encoded characters.

Behavior

JSON and YAML schemas compile when the endpoint is created. A malformed schema fails endpoint creation. The compiled validator stays cached for the lifetime of the endpoint. XSD schemas defer registration to the first validation call. This lets the bridge start in an async context without blocking endpoint creation.

XSD validation delegates to the xml-bridge gRPC backend. The bridge starts as a child process on the first XSD validation. The XsdBridgeBackend caches registered schemas up to schemaCacheMaxEntries and re-seeds them on reconnect. JSON and YAML validation never start the bridge.

Validation failure returns a CamelError to the route's error handler. Empty bodies, when failOnNullBody=true (default), also return an error. Pass failOnNullBody=false to let empty bodies flow through without validation. The same toggle exists for headerName mode via failOnNullHeader. The validator endpoint supports Producers only. Consumer creation returns an error.

Bridge lifecycle

CamelContext::stop() does not clean up the xml-bridge child process. The camel run CLI registers a BridgeCleanup lifecycle service that calls XsdBridgeBackend::shutdown() on stop. Library embedders must retain the value from ValidatorComponent::xsd_bridge_backend() and call shutdown().await when it is Some. This requirement only applies after an XSD route starts the bridge. JSON and YAML routes never start it.

Reference: Validator crate CONTEXT. Example source: examples/validator.

Exec

The exec component runs external system processes from a route. It is producer-only. Each route binds to a named Profile. A Profile is a pre-configured capability bundle: executable, argument policy, environment, working directory, and caps. It is pinned at startup. The component runs with execvp semantics, not a shell. ADR-0037 defines the fail-closed capability model.

The exec-example shows two profiles wired against a timer source:

use camel_api::CamelError;
use camel_builder::{RouteBuilder, StepAccumulator};
use camel_component_api::ComponentBundle;
use camel_component_exec::ExecBundle;
use camel_component_log::LogComponent;
use camel_component_timer::TimerComponent;
use camel_core::context::CamelContext;

fn register_exec_bundle(ctx: &mut CamelContext) {
    let toml_str = r#"
workspace_root = "."

[[profiles]]
name = "echo"
executable = "echo"
args = { allow = "any" }
timeout_secs = 10
working_dir = "."
accepted_exit_codes = [0]

[[profiles]]
name = "date"
executable = "date"
timeout_secs = 5
working_dir = "."
accepted_exit_codes = [0]
"#;
    let value: toml::Value = toml::from_str(toml_str).expect("parse toml");
    let bundle = ExecBundle::from_toml(value).expect("bundle");
    bundle.register_all(ctx);
}

#[tokio::main]
async fn main() -> Result<(), CamelError> {
    let mut ctx = CamelContext::builder().build().await?;
    register_exec_bundle(&mut ctx);
    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:tick?period=500&repeatCount=1")
        .route_id("exec-echo")
        .set_header(
            camel_component_exec::headers::CAMEL_EXEC_ARGS,
            serde_json::json!(["-n", "Hello", "World"]),
        )
        .to("exec:echo")
        .to("log:info?showBody=true&showHeaders=true")
        .build()?;
    ctx.add_route_definition(route).await?;

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

# Camel.toml
[components.exec]
workspace_root = "."

[[components.exec.profiles]]
name = "echo"
executable = "echo"
args = { allow = "any" }
timeout_secs = 10
working_dir = "."
accepted_exit_codes = [0]

# Route
routes:
  - id: exec-echo
    from: "timer:tick?period=500&repeatCount=1"
    steps:
      - setHeader:
          name: CamelExecArgs
          value: ["-n", "Hello", "World"]
      - to: "exec:echo"
      - to: "log:info?showBody=true&showHeaders=true"

URI

exec:{profile-name}

The path segment names the Profile to run. The component verifies the profile exists at endpoint creation time. A missing profile fails route startup, not the first invocation.

Configuration

Config lives under [components.exec] in Camel.toml. The ExecBundle deserializes it and calls ExecGlobalConfig::validate() at startup. Validation pins the canonical executable path, validates every working_dir against the canonical workspace root, and rejects duplicate profile names.

Default deny_env patterns: LD_*, DYLD_*, PYTHONPATH, RUSTFLAGS, GIT_*, SSH_AUTH_SOCK, *_TOKEN, *_KEY. They block library-preload and secret-injection vectors. PATH is opt-in.

Each [[components.exec.profiles]] entry has:

A profile with zero profiles fails startup with no profiles configured (fail-closed: refusing to execute anything). There is no default profile, no allow-all mode, and no shell convenience syntax.

Security model

ADR-0037 fixes eleven decisions. The full text is the authority. The ones that shape every route:

Profile-pinned, fail-closed. A capability lives in a profile declared at startup. Exchange data cannot select an executable, change a path, or modify a policy. The component refuses to start with zero profiles. There is no exec:shell?cmd=... shortcut.

No shell by default. Commands run as binary + literal argv. No string concatenation, no sh -c wrapper. The component rejects known shells (sh, bash, zsh, cmd.exe, pwsh, …) at runtime unless the profile sets allow_shell = true. Even with allow_shell, the binary is the shell itself, called with explicit argv, never a concatenated command string.

Canonical pin at startup. The producer resolves the executable once during validate(). At runtime it uses the pinned path, never a fresh PATH lookup. The pin is not symlink-resolved, because multi-call binaries (BusyBox, uutils) dispatch on argv[0]. Canonicalization would break them.

Per-element argument policy. Every element in CamelExecArgs runs through ArgPolicy. deny_flags is applied first with prefix match. An arg that matches both deny_flags and the allow mode is denied. The default policy is exact { values: [] }, which denies every non-empty arg. The route must opt in to any or specify values.

Empty environment by default. The child starts with no host env. Three layers compose: env.allow (copy from host), env.set (explicit pairs), then global deny_env (strip globs, last and always winning). Operators must allow PATH explicitly for PATH-dependent binaries.

Working-directory confinement. working_dir is validated at startup against the canonical workspace root. Absolute paths fail. Paths containing .. fail. Resolved paths that escape the root fail. The component does not create missing directories. The operator must pre-create them.

No dynamic override from exchange data. The profile is fixed by the endpoint URI. Conditional dispatch between profiles lives in route EIPs (choice, recipient_list), where the route author controls the branching, not the exchange payload. This is the lesson from ADR-0034 (ControlBus).

Argument policy modes

Combine deny_flags = ["--upload-pack"] with args = { allow = "any" } to accept arbitrary args but block a known-dangerous flag. The denylist always wins.

Headers

Input and output headers travel on the Exchange.

The body after a producer call is a JSON ExecResult:

{
  "exit_code": 0,
  "stdout": "aGVsbG8K",
  "stderr": "",
  "stdout_truncated": false,
  "stderr_truncated": false,
  "timed_out": false,
  "profile": "echo",
  "duration_ms": 12
}

stdout and stderr are base64 strings. Raw bytes would make pathological JSON. The CamelExecStderr header is lossy-UTF8 for use inside choice() and log: predicates. The dual representation is intentional.

Non-error outcomes

A timeout or an exit code outside accepted_exit_codes does not return Err. The producer returns Ok(exchange) with the ExecResult body and headers set. This is forced by the Service<Exchange> contract: the Tower trait discards the mutated exchange on Err, and these outcomes carry output the route should see.

Branch on outcome with CamelExecExitAccepted:

- to: "exec:build"
- choice:
    when:
      - predicate: "${header.CamelExecExitAccepted} == true"
        steps:
          - to: "log:info?showBody=true"
      - predicate: "${header.CamelExecTimedOut} == true"
        steps:
          - to: "log:warn?showBody=true"
    otherwise:
      - to: "log:error?showBody=true"

Only pre- and during-spawn failures return Err: arg policy denial, shell rejection, workdir escape, stdin over the cap, and OS spawn errors. Those route to the route's ErrorHandler.