diff --git a/spring-integration-reference/src/aggregator-resequencer.xml b/spring-integration-reference/src/aggregator-resequencer.xml index 13f91a977d..aaeca0363e 100644 --- a/spring-integration-reference/src/aggregator-resequencer.xml +++ b/spring-integration-reference/src/aggregator-resequencer.xml @@ -62,7 +62,7 @@ The Resequencer works in a similar way to the Aggregator, in the sense that it uses the CORRELATION_ID to store messages in groups, the difference being that all what the Resequencer does, is to release them - in the order of their SEQUENCE_NUMBER. + in the order of their SEQUENCE_NUMBER. With respect to that, the user might opt to release all messages at once (after the whole sequence, according to the SEQUENCE_SIZE, has @@ -104,14 +104,14 @@ message sequence to aggregate is complete (and grouping them according to their CORRELATION_ID), and implementing the timeout functionality. The responsibility of deciding whether the message sequence is complete - is delegated to a CompletionStrategy instance. + is delegated to a CompletionStrategy instance. A brief highlight of the base AbstractMessageAggregator (the responsibility of implementing the aggregateMessages method is left to the developer): - public abstract class AbstractMessageAggregator extends AbstractMessageBarrierConsumer { + public abstract class AbstractMessageAggregator extends AbstractMessageBarrierConsumer { private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); @@ -126,18 +126,18 @@ implement the aggregateMessages method. However, there are better suited (which reads, less coupled to the API) solutions for implementing the aggregation logic, which can be configured easily - either through XML or through annotations. + either through XML or through annotations. - In the interest of code simplicity, and promoting good + In the interest of code simplicity, and promoting best practices such as low coupling, testability, etc., the preferred way - of implementing the aggregation logic by implementing a POJO, and + of implementing the aggregation logic is by implementing a POJO, and using the XML or annotation support for setting it up in the application. The CompletionStrategy interface is defined as follows: - public interface CompletionStrategy { + public interface CompletionStrategy { boolean isComplete(List<Message<?>> messages); @@ -145,17 +145,125 @@ Spring Integration provides an out-of-the box implementation for CompletionStrategy, the - SequenceSizeCompletionStrategy. This - implementation uses the SEQUENCE_NUMBER and SEQUENCE_SIZE of the - arriving messages for deciding when a message group is complete and - ready to be aggregated. + SequenceSizeCompletionStrategy This implementation uses the + SEQUENCE_NUMBER and SEQUENCE_SIZE of the arriving messages for deciding + when a message group is complete and ready to be + aggregated.
Configuration using XML - +
+ Configuring an aggregator through XML + + 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: + + <aggregator id="completelyDefinedAggregator" + input-channel="completelyDefinedAggregatorInput" + output-channel="outputChannel" + discard-channel="discardChannel" + ref="aggregatorBean" + method="add" + completion-strategy="completionStrategy" + completion-strategy-method="checkCompleteness" + timeout="42" + send-partial-result-on-timeout="true" + reaper-interval="135" + tracked-correlation-id-capacity="99" + send-timeout="86420000" /> + + + + The id of the aggregator is + optional. + + + + The input channel of the aggregator. + Required. + + + + The channel where the aggregator will send the aggregation + results. Required. + + + + The channel where the aggregator will send the messages that + timed out (if send-partial-results-on-timeout is + true. Optional. + + + + A reference to a bean defined in the application context. The + bean must either extend AbstractMessageAggregator or be + a POJO. In the latter case the method attribute must be + defined as well. Required. + + + + A method defined on the bean referenced by ref, + that implements the message aggregation + algorithm. Optional, with restrictions (see + above). + + + + 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-mTethod attribute must be + defined as well. Optional. + + + + A method defined on the bean referenced by + completion-strategy, that + implements the completion decision algorithm. Optional, + with restrictions (see above). Optional. + + + + 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. + + +
+ +
+ Configuring a resequencer by using XML + + +