Aggregator

Aggregator

The Aggregator from the EIP patterns allows you to combine a number of messages together into a single message.

A correlation Expression is used to determine the messages which should be aggregated together. If you want to aggregate all messages into a single message, just use a constant expression. An AggregationStrategy is used to combine all the message exchanges for a single correlation key into a single message exchange. The default strategy just chooses the latest message; so its ideal for throttling messages.

For example, imagine a stock market data system; you are receiving 30,000 messages per second; you may want to throttle down the updates as, say, a GUI cannot cope with such massive update rates. So you may want to aggregate these messages together so that within a window (defined by a maximum number of messages or a timeout), messages for the same stock are aggregated together; by just choosing the latest message and discarding the older prices. (You could apply a delta processing algorithm if you prefer to capture some of the history).

Using the Fluent Builders

The following example shows how to aggregate messages so that only the latest message for a specific value of the cheese header are sent.

If you were using JMS then you may wish to use the JMSDestination header as the correlation key; or some custom header for the stock symbol (using the above stock market example).

from("activemq:someReallyFastTopic").aggregator(header("JMSDestination")).to("activemq:someSlowTopicForGuis");

You can of course use many different Expression languages such as XPath, XQuery, SQL or various Scripting Languages.
Here is an example using XPath:

//aggregate based on the message content using an XPath expression
//example assumes an XML document starting with <stockQuote symbol='APACHE'>
from("seda:start").aggregate().xpath("name(/stockQuote[@symbol='APACHE'])", String.class).batchSize(5).to("mock:result");

For further examples of this pattern in use you could look at the junit test case

Using the Spring XML Extensions

<aggregator>
  <simple>header.cheese</simple>
  <to uri="mock:result"/>
</aggregator>

The following example shows how to create a simple aggregator using the XML notation; using an Expression for the correlation value used to aggregate messages together.

You can specify your own AggregationStrategy if you prefer as shown in the following example

Notice how the strategyRef attribute is used on the <aggregator> element to refer to the custom strategy in Spring.

Batch options

The aggregator supports the following batch options:

AggregationCollection and AggregationStrategy

This aggregator uses a AggregationCollection to store the exchanges that is currently aggregated. The AggregationCollection uses a correlation Expression and an AggregationStrategy.

• The correlation Expression is used to correlate the incoming exchanges. The default implementation will group messages based on the correlation expression. Other implementations could for instance just add all exchanges as a batch.
• The strategy is used for aggregate the old (lookup by its correlation id) and the new exchanges together into a single exchange. Possible implementations include performing some kind of combining or delta processing, such as adding line items together into an invoice or just using the newest exchange and removing old exchanges such as for state tracking or market data prices; where old values are of little use.

Camel provides these implementations:

• DefaultAggregationCollection
• PredicateAggregationCollection
• UseLatestAggregationStrategy

Examples

Default example

By default Camel uses DefaultAggregationCollection and UseLatestAggregationStrategy, so this simple example will just keep the latest received exchange for the given correlation Expression:

Using PredicateAggregationCollection

The PredicateAggregationCollection is an extension to DefaultAggregationCollection that uses a Predicate as well to determine the completion. For instance the Predicate can test for a special header value, a number of maximum aggregated so far etc. To use this the routing is a bit more complex as we need to create our AggregationCollection object as follows:

In this sample we use the predicate that we want at most 3 exchanges aggregated by the same correlation id, this is defined as:

header(Exchange.AGGREGATED_COUNT).isEqualTo(3)

Using this the aggregator will complete if we receive 3 exchanges with the same correlation id or when the specified timeout of 500 msecs has elapsed (whichever criteria is met first).

Using custom aggregation strategy

In this example we will aggregate incoming bids and want to aggregate the highest bid. So we provide our own strategy where we implement the code logic:

Then we setup the routing as follows:

And since this is based on an unit test we show the test code that send the bids and what is expected as the winners:

Using custom aggregation collection

In this example we will aggregate incoming bids and want to aggregate the bids in reverse order (this is just an example). So we provide our own collection where we implement the code logic:

Then we setup the routing as follows:

And since this is based on an unit test we show the test code that send the bids and what is expected as the expected reverse order:

Custom aggregation collection in Spring DSL

You can also specify a custom aggregation collection in the Spring DSL. Here is an example for Camel 2.0

In Camel 1.5.1 you will need to specify the aggregator as

<aggregator batchTimeout="500" collectionRef="aggregatorCollection">
  <expression/>
  <to uri="mock:result"/>
</aggregator>

See also

• The Loan Broker Example which uses an aggregator