Spring Integration supports the configuration of an aggregator via XML through the <aggregator/> element. Below you can see an example of an aggregator with all optional parameters defined.
<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"
correlation-strategy="correlationStrategyBean"
correlation-strategy-method="correlationStrategyMethod"
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 will use sequence size) . |
| A method defined on the bean referenced by
|
| A reference to a bean that implements the correlation strategy. The bean can be an implementation of the CorrelationStrategy interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use the correlation id header attribute) . |
| 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.
An implementation of the correlation strategy bean for the example above may be as follows:
public class PojoCorrelationStrategy { ... public Long groupsNumbersByLastDigit(Long number) { return number % 10; } }
For example, this aggregator would group numbers by some criterion (in our case the remainder by dividing to 10) and will hold on the group until the sum of the numbers which represents the payload exceeds a certain value.
Wherever it makes sense, the completion strategy method, correlation strategy method and the aggregator method can be combined in a single bean (all of them or any two).