diff --git a/spring-integration-reference/src/aggregator-resequencer.xml b/spring-integration-reference/src/aggregator-resequencer.xml
index 394157d706..d21dd30e80 100644
--- a/spring-integration-reference/src/aggregator-resequencer.xml
+++ b/spring-integration-reference/src/aggregator-resequencer.xml
@@ -128,6 +128,33 @@
implementing the aggregation logic, which can be configured easily
either through XML or through annotations.
+ In general, any ordinary Java class (i.e. POJO) can implement the
+ aggregation algorithm. For doing so, it must provide a method that
+ accepts as an argument a single java.util.List (parametrized lists are
+ supported as well). This method will be invoked for aggregating
+ messages, as follows:
+
+
+
+ if the argument is a parametrized java.util.List, and the
+ parameter type is assignable to Message, then the whole list of
+ messages accumulated for aggregation will be sent to the aggregator
+
+
+
+
+ if the argument is a non-parametrized java.util.List or the
+ parameter type is not assignable to Message, then the method will
+ receive the payloads of the accumulated messages
+
+
+
+ if the return type is not assignable to Message, then it will
+ be treated as the payload for a Message that will be created
+ automatically by the framework.
+
+
+
In the interest of code simplicity, and promoting best
practices such as low coupling, testability, etc., the preferred way
@@ -143,6 +170,32 @@
}
+ In general, any ordinary Java class (i.e. POJO) can implement the
+ completion decision mechanism. For doing so, it must provide a method
+ that accepts as an argument a single java.util.List (parametrized lists
+ are supported as well), and returns a boolean value. This method will be
+ invoked after the arrival of a new message, to decide whether the group
+ is complete or not, as follows:
+
+
+
+ if the argument is a parametrized java.util.List, and the
+ parameter type is assignable to Message, then the whole list of
+ messages accumulated in the group will be sent to the method
+
+
+
+ if the argument is a non-parametrized java.util.List or the
+ parameter type is not assignable to Message, then the method will
+ receive the payloads of the accumulated messages
+
+
+
+ the method must return true if the message group is complete
+ and ready for aggregation, and false otherwise.
+
+
+
Spring Integration provides an out-of-the box implementation for
CompletionStrategy, the
SequenceSizeCompletionStrategy This implementation uses the
@@ -163,20 +216,27 @@
on how to define such an element is presented below, as well as
it:
- <aggregator id="completelyDefinedAggregator"
- input-channel="completelyDefinedAggregatorInput"
+ <channel id="inputChannel"/>
+
+<aggregator id="completelyDefinedAggregator"
+ input-channel="inputChannel"
output-channel="outputChannel"
discard-channel="discardChannel"
ref="aggregatorBean"
method="add"
- completion-strategy="completionStrategy"
+ 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" />
+ send-timeout="86420000" />
+
+<channel id="outputChannel"/>
+
+<bean id="aggregatorBean" class="sample.PojoAggregator"/>
+
+<bean id="completionStrategyBean" class="sample.PojoCompletionStrategy"/>
@@ -191,13 +251,14 @@
The channel where the aggregator will send the aggregation
- results. Required.
+ results. Optional (not required, because the aggregator
+ will honor .
The channel where the aggregator will send the messages that
timed out (if send-partial-results-on-timeout is
- true. Optional.
+ false). Optional.
@@ -219,19 +280,21 @@
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-mTethod attribute must be
- defined as well. Optional.
+ defined as well. Optional (by default, the aggregator
+ .
A method defined on the bean referenced by
completion-strategy, that
implements the completion decision algorithm. Optional,
- with restrictions (see above). Optional.
+ with restrictions (requires completion-strategy to be
+ present).
The timeout for aggregating messages (counted from the arrival
- of the first message). Optional.
+ of the first message). Optional.
@@ -243,32 +306,163 @@
The interval (in milliseconds) at which a reaper task is
executed, checking if there are any timed out groups.
- Optional.
+ 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.
+ processed (aggregated or discarded).
+ Optional.
- The timeout for sending out messages. 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.
Configuring a resequencer by using XML
-
+ Configuring a resequencer requires only including the appropriate
+ element in XML.
+
+ A sample resequencer configuration is shown below.
+
+ <channel id="inputChannel"/>
+
+<channel id="outputChannel"/>
+
+<resequencer id="completelyDefinedResequencer"
+ input-channel="inputChannel"
+ output-channel="outputChannel"
+ discard-channel="discardChannel"
+ release-partial-sequences="true"
+ timeout="42"
+ send-partial-result-on-timeout="true"
+ reaper-interval="135"
+ tracked-correlation-id-capacity="99"
+ send-timeout="86420000" />
+
+
+
+ The id of the resequencer is
+ optional.
+
+
+
+ The input channel of the resequencer.
+ Required.
+
+
+
+ The channel where the resequencer will send the reordered
+ messages. Optional.
+
+
+
+ The channel where the resequencer will send the messages
+ that timed out (if send-partial-result-on-timeout is
+ false). Optional.
+
+
+
+ Whether to send out ordered sequences as soon as they are
+ available, or only after the whole message group arrives.
+ Optional (true by default).
+
+
+
+ The timeout for reordering message sequences (counted from
+ the arrival of the first message).
+ Optional.
+
+
+
+ Whether, upon the expiration of the timeout, the ordered
+ group shall be sent out (even if some of the messages are
+ missing). 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 (reordered or discarded).
+ Optional.
+
+
+
+ The timeout for sending out messages.
+ Optional.
+
+
Configuration using annotations
-
+ This section will cover only the configuration for aggregators.
+ Since there is no custom behaviour to be implemented in Java classes for
+ resequencers, there is no annotation support for it.
+
+ An aggregator configured using annotations can look like
+ this.
+
+ public class Waiter {
+ ...
+
+ @Aggregator
+ public Delivery aggregatingMethod(List<OrderItem> items) {
+ ...
+ }
+
+ @CompletionStrategy
+ public boolean completionChecker(List<Message<?>> messages) {
+ ...
+ }
+
+}
+