Aggregator Pattern

Five workers each paint one piece of a mural on separate canvases. Someone needs to take all five canvases and arrange them into the final picture. They need to know which pieces belong together, what order they go in, and when all pieces have arrived.

That's the aggregator. It collects related messages from different sources, determines when the set is complete, and combines them into a single output message. It's the natural partner to the splitter: what the splitter breaks apart, the aggregator reassembles.

But aggregators aren't just for reassembly. They also combine messages that were never a single message to begin with. A loan approval system collects responses from the credit check service, the income verification service, and the identity verification service. Three independent services producing three independent messages about the same loan application. The aggregator collects all three and produces a single "loan decision" message.

Apache Camel's aggregator processes millions of groups per hour at organizations like the UK National Health Service. AWS Step Functions uses the same pattern in its Map state, fan out to parallel branches and aggregate the results. Spring Integration's aggregator component handles correlation, completeness, and timeout out of the box.

1. Correlation: Which messages belong together? The aggregator groups messages by a correlation ID, typically a business identifier like orderId, applicationId, or requestId. All messages with the same correlation ID are part of the same group.

2. Completeness: When have all messages arrived?

3. Aggregation: How are messages combined?

function aggregate(messages: Message[]): AggregatedResult {
  // Sort by sequence number
  const sorted = messages.sort((a, b) => a.sequenceNumber - b.sequenceNumber);
  
  // Combine into final result
  return {
    orderId: sorted[0].orderId,
    lineItems: sorted.map(m => m.lineItem),
    allShipped: sorted.every(m => m.status === 'shipped'),
    totalCost: sorted.reduce((sum, m) => sum + m.cost, 0)
  };
}

State management. The aggregator must store partial groups while waiting for remaining messages. This state can be in memory (fast but lost on crash) or in a database (durable but slower). For production systems, use a database. Aggregator crashes shouldn't lose partially collected groups.