diff --git a/spring-integration-reference/src/aggregator.xml b/spring-integration-reference/src/aggregator.xml index 764039bed4..483ef00486 100644 --- a/spring-integration-reference/src/aggregator.xml +++ b/spring-integration-reference/src/aggregator.xml @@ -14,10 +14,7 @@ Technically, the Aggregator is more complex than a Splitter, because it is required to maintain state (the Messages to-be-aggregated), to - decide when the complete group of Messages is available, and to timeout if - necessary. Furthermore, in case of a timeout, the Aggregator needs to know - whether to send the partial results or to discard them to a separate - channel. + decide when the complete group of Messages is available.
@@ -28,12 +25,7 @@ Aggregator will create a single message by processing the whole group, and will send that aggregated message as output. - As messages might arrive with a certain delay (or certain messages - from the group might not arrive at all), the Aggregator can specify a - timeout (counted from the moment when the first message in the group has - arrived), and whether, in the case of a timeout, the group should be - discarded, or the Aggregator should merely attempt to create a single - message out of what has arrived so far. An important aspect of + An important aspect of implementing an Aggregator is providing the logic that has to be executed when the aggregation (creation of a single message out of many) takes place. @@ -45,9 +37,9 @@ specifying how the messages should be grouped together, by using a CorrelationStrategy (see below). - An important concern with respect to the timeout is, what happens if + Another important concern is, what happens if late messages arrive after the aggregation has taken place? In this case, - a configuration option allows the user to decide whether they should be + the user needs to be able to decide whether they should be discarded or not.
@@ -58,13 +50,14 @@ - The base class AbstractMessageAggregator and its - subclass MethodInvokingMessageAggregator + The interface MessageGroupProcessor and related + base class AbstractAggregatingMessageGroupProcessor and its + subclass MethodInvokingAggregatingMessageGroupProcessor - The CompletionStrategy interface and its default - implementation SequenceSizeCompletionStrategy + The ReleaseStrategy interface and its default + implementation SequenceSizeReleaseStrategy @@ -74,11 +67,11 @@
- AbstractMessageAggregator + CorrelatingMessageHandler - The AbstractMessageAggregator is a + The CorrelatingMessageHandler is a MessageHandler implementation, encapsulating the common - functionalities of an Aggregator, which are: + functionalities of an Aggregator (and other correlating use cases), which are: correlating messages into a group to be aggregated @@ -93,34 +86,35 @@ processing the completed group into a single aggregated message - recognizing and responding to a timed-out completion attempt + recognizing and responding to an expired group The responsibility of deciding how the messages should be grouped together is delegated to a CorrelationStrategy instance. The responsibility - of deciding whether the message group is complete is delegated to a - CompletionStrategy instance. + of deciding whether the message group can be released is delegated to a + ReleaseStrategy instance. Here is a brief highlight of the base - AbstractMessageAggregator (the responsibility of + AbstractAggregatingMessageGroupProcessor (the responsibility of implementing the aggregateMessages method is left to the developer): - public abstract class AbstractMessageAggregator - extends AbstractMessageBarrierHandler { + public abstract class AbstractAggregatingMessageGroupProcessor + implements MessageGroupProcessor { - private volatile CompletionStrategy completionStrategy - = new SequenceSizeCompletionStrategy(); + protected Map<String, Object> aggregateHeaders(MessageGroup group) { .... + } - protected abstract Message<?> aggregateMessages(List<Message<?>> messages); + protected abstract Object aggregatePayloads(MessageGroup group); } - It also inherits the following default CorrelationStrategy: + The CorrelationStrategy is owned by the CorrelatingMessageHandler and it has + a default value based on the correlation ID message header: private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID); - When appropriate, the simplest option is the DefaultMessageAggregator. + When appropriate, the simplest option is the DefaultAggregatingMessageGroupProcessor. It creates a single Message whose payload is a List of the payloads received for a given group. It uses the default CorrelationStrategy and CompletionStrategy as shown above. This works well for simple @@ -138,8 +132,8 @@ When implementing a specific aggregator object for an application, - a developer can extend AbstractMessageAggregator and - implement the aggregateMessages method. However, there are + a developer can extend AbstractAggregatingMessageGroupProcessor and + implement the aggregatePayloads 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. @@ -180,14 +174,14 @@
- CompletionStrategy + ReleaseStrategy - The CompletionStrategy interface is defined as + The ReleaseStrategy interface is defined as follows: - public interface CompletionStrategy { + public interface ReleaseStrategy { - boolean isComplete(List<Message<?>> messages); + boolean canRelease(MessageGroup messages); } @@ -212,17 +206,27 @@ - the method must return true if the message group is complete - and ready for aggregation, and false otherwise. + the method must return true if the message group is ready + for aggregation, and false otherwise. + When the group is released for aggregation, all its + unmarked messages are processed and then marked so they will not + be processed again. If the group is also complete (i.e. if all + messages from a sequence have arrived or if there is no sequence + defined) then the group is removed from the message store. + Partial sequences can be released, in which case the next time + the ReleaseStrategy is called it will be presented + with a group containing marked messages (already processed) and + unmarked messages (a potential new partial sequence) + Spring Integration provides an out-of-the box implementation for - CompletionStrategy, the - SequenceSizeCompletionStrategy. This implementation uses + ReleaseStrategy, the + SequenceSizerReleaseStrategy. 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. As shown above, it is also the default strategy. + aggregated. As shown above, it is also the default strategy.
@@ -274,23 +278,21 @@ discard-channel="discardChannel" ref="aggregatorBean" method="add" - completion-strategy="completionStrategyBean" - completion-strategy-method="checkCompleteness" + release-strategy="releaseStrategyBean" + release-strategy-method="canRelease" correlation-strategy="correlationStrategyBean" correlation-strategy-method="groupNumbersByLastDigit" - timeout="42" - send-partial-result-on-timeout="true" - reaper-interval="135" - tracked-correlation-id-capacity="99" - send-timeout="86420000" /> + message-store="messageStore" + send-partial-result-on-expiry="true" + send-timeout="86420000" /> <channel id="outputChannel"/> <bean id="aggregatorBean" class="sample.PojoAggregator"/> -<bean id="completionStrategyBean" class="sample.PojoCompletionStrategy"/> +<bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/> <bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/> @@ -341,7 +343,7 @@ A method defined on the bean referenced by - completion-strategy, that implements the + release-strategy, that implements the completion decision algorithm. Optional, with restrictions (requires completion-strategy to be present). @@ -364,34 +366,21 @@ present). - - The timeout (in milliseconds) for aggregating messages (counted - from the arrival of the first message). Optional. - - + + A reference to a MessageGroupStore that + can be used to store groups of messages under their + correlation key until they are + complete. Optional with default a + volatile in-memory store. + - - Whether upon the expiration of the timeout, the aggregator shall + + Whether upon the expiration of the message group, the aggregator will try to aggregate the messages that have already arrived. 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). Set this value to 0 if you - do not want the messages to be discarded in such a scenario. - Optional. - - - + The timeout for sending the aggregated messages to the output or reply channel. Optional. @@ -432,9 +421,9 @@ An implementation of the completion strategy bean for the example above may be as follows: - public class PojoCompletionStrategy { + public class PojoReleaseStrategy { ... - public boolean checkCompleteness(List<Long> numbers) { + public boolean canRelease(List<Long> numbers) { int sum = 0; for (long number: numbers) { sum += number; @@ -444,7 +433,7 @@ } - Wherever it makes sense, the completion strategy method and + Wherever it makes sense, the release strategy method and the aggregator method can be combined in a single bean. @@ -465,7 +454,7 @@ certain value. - Wherever it makes sense, the completion strategy method, correlation + Wherever it makes sense, the release strategy method, correlation strategy method and the aggregator method can be combined in a single bean (all of them or any two). @@ -485,8 +474,8 @@ ... } - @CompletionStrategy - public boolean completionChecker(List<Message<?>> messages) { + @ReleaseStrategy + public boolean releaseChecker(List<Message<?>> messages) { ... } @@ -506,7 +495,7 @@ An annotation indicating that this method shall be - used as the completion strategy of an aggregator. If not present on + used as the release strategy of an aggregator. If not present on any method, the aggregator will use the SequenceSizeCompletionStrategy. diff --git a/spring-integration-reference/src/resequencer.xml b/spring-integration-reference/src/resequencer.xml index ed21f41a10..8a59ea0b02 100644 --- a/spring-integration-reference/src/resequencer.xml +++ b/spring-integration-reference/src/resequencer.xml @@ -22,10 +22,7 @@ With respect to that, the user might opt to release all messages at once (after the whole sequence, according to the SEQUENCE_SIZE, has been - released), or as soon as a valid sequence is available. Another option is - to set a timeout, deciding whether to drop the whole sequence if the - timeout has expired, and not all messages have arrived, or to release the - messages accumulated so far, in the appropriate order. + released), or as soon as a valid sequence is available.
@@ -45,10 +42,8 @@ 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" + message-store="messageStore" + send-partial-result-on-expiry="true" send-timeout="86420000" /> @@ -76,34 +71,31 @@ Whether to send out ordered sequences as soon as they are available, or only after the whole message group arrives. - Optional (true by default). + Optional (false by default). If this + flag is not specified (so a complete sequence is defined by the sequence + headers) then it can make sense to provide a custom Comparator + to be used to order the messages when sending + (use the XML attribute comparator to point to a bean + definition). If release-partial-sequences is true then + there is no way with a custom comparator to define a partial sequence. To + do that you would have to provide a release-strategy (also + a reference to another bean definition, either a POJO or a ReleaseStrategy). - The timeout (in milliseconds) for reordering message sequences (counted from the - arrival of the first message). Optional. + A reference to a MessageGroupStore that + can be used to store groups of messages under their + correlation key until they are + complete. Optional with default a + volatile in-memory store. - Whether, upon the expiration of the timeout, the ordered group - shall be sent out (even if some of the messages are missing). + Whether, upon the expiration of the group, the ordered group + should 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.