This applies for Camel version 2.3 or newer. If you use an older version then use this Aggregator link instead.
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 aggregator supports the following options:
Mandatory Expression which evaluates the correlation key to use for aggregation. The Exchange which has the same correlation key is aggregated together. If the correlation key could not be evaluated an Exception is thrown. You can disable this by using the
From Camel 2.9.2 onwards the strategy can also be a
From Camel 2.16: the strategy can also be a
A reference to lookup the
Camel 2.12: This option can be used to explicit declare the method name to use, when using POJOs as the
Camel 2.12: If this option is
Number of messages aggregated before the aggregation is complete. This option can be set as either a fixed value or using an Expression which allows you to evaluate a size dynamically - will use
Time in millis that an aggregated exchange should be inactive before its complete. This option can be set as either a fixed value or using an Expression which allows you to evaluate a timeout dynamically - will use
A repeating period in millis by which the aggregator will complete all current aggregated exchanges. Camel has a background task which is triggered every period. You cannot use this option together with
A Predicate to indicate when an aggregated exchange is complete.
From Camel 2.15: if this is not specified and the
This option is if the exchanges are coming from a Batch Consumer. Then when enabled the Aggregator2 will use the batch size determined by the Batch Consumer in the message header
Camel 2.9 Indicates to complete all current aggregated exchanges when the context is stopped
|Camel 2.16: Indicates to wait to complete all current and partial (pending) aggregated exchanges when the context is stopped. This also means that we will wait for all pending exchanges which are stored in the aggregation repository to complete so the repository is empty before we can stop. You may want to enable this when using the memory based aggregation repository that is memory based only, and do not store data on disk. When this option is enabled, then the aggregator is waiting to complete all those exchanges before its stopped, when stopping CamelContext or the route using it.|
Whether or not to eager check for completion when a new incoming Exchange has been received. This option influences the behavior of the
If enabled then Camel will group all aggregated Exchanges into a single combined
Note: this option does not support persistent repository with the aggregator. See further below for an example and more details.
Whether or not to ignore correlation keys which could not be evaluated to a value. By default Camel will throw an Exception, but you can enable this option and ignore the situation instead.
Whether or not too late Exchanges should be accepted or not. You can enable this to indicate that if a correlation key has already been completed, then any new exchanges with the same correlation key be denied. Camel will then throw a
Camel 2.5: Whether or not exchanges which complete due to a timeout should be discarded. If enabled then when a timeout occurs the aggregated message will not be sent out but dropped (discarded).
Allows you to plugin you own implementation of
Reference to lookup a
When aggregated are completed they are being send out of the aggregator. This option indicates whether or not Camel should use a thread pool with multiple threads for concurrency. If no custom thread pool has been specified then Camel creates a default pool with 10 concurrent threads.
Reference to lookup a
Camel 2.9: If using either of the
Camel 2.9: Reference to lookup a
Camel 2.11: Turns on using optimistic locking, which requires the
Camel 2.11.1: Allows to configure retry settings when using optimistic locking.
The following properties are set on each aggregated Exchange:
The total number of Exchanges aggregated into this combined Exchange.
Indicator how the aggregation was completed as a value of either:
AggregationStrategy is used for aggregating 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.
Notice the aggregation strategy is a mandatory option and must be provided to the aggregator.
Here are a few example
AggregationStrategy implementations that should help you create your own custom strategy.