Aggregate¶

Aggregate implements the Aggregator EIP: it buffers incoming exchanges by a correlation key and emits a single combined exchange when a completion condition is met. The file also defines the built-in AggregationStrategy implementations used by aggregate, split, multicast, enrich, and scatter-gather.

Processors in this file¶

NewAggregate¶

YAML form:

steps:
  - aggregate:
      correlation: header("order-id")
      strategy: list
      completion-size: 5
      completion-timeout: 10s
      steps:
        - target: kafka://order-batches

Constructor:

NewAggregate(
    correlation engine.Expression,
    strategy engine.AggregationStrategy,
    completionSize int,
    completionTimeout time.Duration,
    processor engine.Processor,
) *Aggregate

How it works:

1. Evaluates CorrelationExpression against the incoming exchange. Stringifies the result (fmt.Sprintf("%v", val)) to form the bucket key.
2. Acquires the aggregate's mutex and looks up the buffer for that key. Creates a new buffer if none exists, and arms a time.AfterFunc(CompletionTimeout, ...) if CompletionTimeout > 0. (The timer uses context.Background() so it fires even if the original request context is cancelled.)
3. Appends a copy of the exchange to the buffer's exchange list.
4. Calls Strategy.Aggregate(ctx, buf.aggregate, ex) to fold the new exchange into the running aggregate.
5. Checks completion-size: if the buffer now contains ≥ CompletionSize exchanges, completion fires.
6. On completion: removes the buffer, stops the timer, sets HelixAggregatedSize and HelixAggregatedCorrelationKey on the aggregate exchange, and runs the downstream Processor.

The aggregate implements engine.StoppableProcessor via Stop(), which cancels all pending completion timers — useful for tests and graceful shutdown.

Correlation expression cheat sheet¶

The correlation: field is a Starlang expression. Common shapes:

Anything stringifiable works — ints, floats, bools, maps. Two exchanges land in the same bucket iff fmt.Sprintf("%v", x) == fmt.Sprintf("%v", y).

Aggregation strategies¶

The file exposes five built-in strategies. Selected by name via AggregationStrategyByName from YAML.

ListAggregationStrategy — list¶

Appends every incoming body into a []any. The aggregated exchange's body is the accumulating list. Use for classic scatter-gather (multicast + list = collect N responses).

LatestAggregationStrategy — latest¶

Keeps only the most recent exchange. Useful when only the final state of a chain matters.

FirstAggregationStrategy — first¶

Keeps only the first exchange. Useful for "fastest-responder" or race-style patterns.

UseOriginalStrategy — useOriginal, use-original, original¶

Discards all responses; the input passes through unchanged. Use this for fire-and-forget fan-out where you don't care about replies (multicast + useOriginal is a broadcast).

MapAggregationStrategy — map¶

Collects responses into a map[string]any keyed by the HelixMulticastIndex (or split index) property. Use when you need to know which processor produced which response.

AggregationStrategyByName¶

Helper that maps a name to a strategy instance:

Used by the reifier to look up the YAML strategy: field. To register a custom strategy, implement engine.AggregationStrategy in code and pass it to the constructor directly.

Pitfalls¶

• CompletionSize and CompletionTimeout can both be set — whichever fires first wins.
• A buffer with no completion configured (both fields zero) will never flush, growing forever. The reifier rejects such configs, but be aware when building in Go.
• The aggregate exchange returned to Processor is a fresh exchange built by the strategy — properties on individual inputs are not merged unless the strategy does it.
• The completion timer captures context.Background() so the downstream Processor runs even after the original request context is cancelled. If your downstream uses ctx, derive a fresh one.

Examples¶

Batch up to 100 events per customer with a 5s flush:

workflows:
  - id: batch-customer-events
    trigger: { uri: kafka://events?group=batcher }
    steps:
      - aggregate:
          correlation: body().customerId
          strategy: list
          completion-size: 100
          completion-timeout: 5s
          steps:
            - log: { message: "flushing customer batch" }
            - target: postgres://customer-events

Source¶

pkg/processor/aggregate.go

See also¶

• multicast — broadcast + aggregate; uses the same strategies.
• split — opposite direction; splits a slice into per-element exchanges.
• routing — Enrich uses aggregation strategies to merge enrichment data.
• eip_advanced — combined multicast + aggregate with timeout.