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