Spring Integration supports the configuration of an aggregator via XML through the <aggregator/> element. A completely defined sample on how to define such an element is presented below, as well as it:
<channel id="inputChannel"/> <aggregator id="completelyDefinedAggregator"input-channel="inputChannel"
output-channel="outputChannel"
discard-channel="discardChannel"
ref="aggregatorBean"
method="add"
completion-strategy="completionStrategyBean"
completion-strategy-method="checkCompleteness"
timeout="42"
send-partial-result-on-timeout="true"
reaper-interval="135"
tracked-correlation-id-capacity="99"
send-timeout="86420000"
/> <channel id="outputChannel"/> <bean id="aggregatorBean" class="sample.PojoAggregator"/> <bean id="completionStrategyBean" class="sample.PojoCompletionStrategy"/>
| The id of the aggregator is optional. |
| The input channel of the aggregator. Required. |
| The channel where the aggregator will send the aggregation results. Optional (because incoming messages can specify a reply channel themselves). |
| The channel where the aggregator will send the messages that
timed out (if |
| A reference to a bean defined in the application context. The bean must implement the aggregation logic as described above. Required. |
| A method defined on the bean referenced by |
| A reference to a bean that implements the decision algorithm as to whether a given message group is complete. The bean can be an implementation of the CompletionStrategy interface or a POJO. In the latter case the completion-strategy-method attribute must be defined as well. Optional (by default, the aggregator . |
| A method defined on the bean referenced by
|
| The timeout for aggregating messages (counted from the arrival of the first message). Optional. |
| Whether upon the expiration of the timeout, the aggregator shall try to aggregate the already arrived messages. Optional (false by default). |
| The interval (in milliseconds) at which a reaper task is executed, checking if there are any timed out groups. Optional. |
| The capacity of the correlation id tracker. Remembers the already processed correlation ids, preventing the formation of new groups for messages that arrive after their group has been already processed (aggregated or discarded). Optional. |
| The timeout for sending out messages. Optional. |
An implementation of the aggregator bean, for example, looks as follows:
public class PojoAggregator { public Long add(List<Long> results) { long total = 0l; for (long partialResult: results) { total += partialResult; } return total; } }
An implementation of the completion strategy bean for the example above may be as follows:
public class PojoCompletionStrategy { ... public boolean checkCompleteness(List<Long> numbers) { int sum = 0; for (long number: numbers) { sum += number; } return sum >= maxValue; } }
Wherever it makes sense, the completion strategy method and the aggregator method can be combined in a single bean.