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.