diff --git a/spring-integration-reference/src/aggregator.xml b/spring-integration-reference/src/aggregator.xml index bac7b66abd..b612af512d 100644 --- a/spring-integration-reference/src/aggregator.xml +++ b/spring-integration-reference/src/aggregator.xml @@ -8,7 +8,7 @@ Introduction Basically a mirror-image of the Splitter, the Aggregator is a type - of Message Consumer that receives multiple Messages and combines them into + of Message Handler that receives multiple Messages and combines them into a single Message. In fact, Aggregators are often downstream consumers in a pipeline that includes a Splitter. @@ -38,9 +38,12 @@ when the aggregation (creation of a single message out of many) takes place. - In Spring Integration, the grouping of the messages for Aggregation - is done based on their CORRELATION_ID message header (i.e. the messages - with the same CORRELATION_ID will be grouped together). + In Spring Integration, the grouping of the messages for aggregation + is done by default based on their CORRELATION_ID message header (i.e. the + messages with the same CORRELATION_ID will be grouped together). However, + this can be customized, and the users can opt for different other ways of + 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 late messages arrive after the aggregation has taken place? In this case, @@ -58,29 +61,38 @@ The base class AbstractMessageAggregator and its subclass MethodInvokingMessageAggregator - - The CompletionStrategy interface and its default implementation SequenceSizeCompletionStrategy + + + The CorrelationStrategy interface and its default + implementation HeaderAttributeCorrelationStrategy + - The AbstractMessageAggregator is a - MessageConsumer implementation, encapsulating the common - functionalities of an Aggregator, which are: storing messages until the - 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. +
+ AbstractMessageAggregator - A brief highlight of the base AbstractMessageAggregator - (the responsibility of implementing the aggregateMessages method is left - to the developer): + The AbstractMessageAggregator is a + MessageHandler implementation, encapsulating the common + functionalities of an Aggregator, which are: storing messages until the + message sequence to aggregate is complete and processing them + afterwards, and implementing the timeout functionality. The + responsibility of deciding how the messages should be grouped together + is delegated to a CorrelationStrategy instance. The responsibility of + deciding whether the message sequence is complete is delegated to a + CompletionStrategy instance. - public abstract class AbstractMessageAggregator - extends AbstractMessageBarrierConsumer { + A brief highlight of the base + AbstractMessageAggregator (the responsibility of + implementing the aggregateMessages method is left to the + developer): + + public abstract class AbstractMessageAggregator + extends AbstractMessageBarrierHandler { private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); @@ -90,86 +102,127 @@ } - For implementing a specific aggregator object for an application, a - developer can extend AbstractMessageAggregator and 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. + For implementing a specific aggregator object for an application, + a developer can extend AbstractMessageAggregator and + 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. - 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: + 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 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 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. - - + + 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 of implementing the aggregation logic is through 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 { +
+ CompletionStrategy + + The CompletionStrategy interface is defined as + follows: + + public interface CompletionStrategy { boolean isComplete(List<Message<?>> messages); } - 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: + 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 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 - + + 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. - - + + 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 - SEQUENCE_NUMBER and SEQUENCE_SIZE of the arriving messages for deciding - when a message group is complete and ready to be - aggregated. + 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. +
+ +
+ CorrelationStrategy + + The CorrelationStrategy interface is defined as + follows: + + public interface CorrelationStrategy { + + Object getCorrelationKey(Message<?> message); + +} + + The method shall return an Object which represents the correlation + key used for grouping messages together. The key must satisfy the + criteria used for a key in a Map with respect to the implementation of + equals() and hashCode(). + + In general, any ordinary Java class (i.e. POJO) can implement the + correlation decision mechanism, and the rules for mapping a message to + method's argument (or arguments) are the same as for a + ServiceActivator (including support for @Header + annotations). The method must return a value, and the value must not be + null. + + Spring Integration provides an out-of-the box implementation for + CorrelationStrategy, the + HeaderAttributeCorrelationStrategy. + This implementation returns the value of one of the message headers + (whose name is specified by a constructor argument) as the correlation + key. By default, the correlation strategy is a + HeaderAttributeCorrelationStrategy returning the value of the + CORRELATION_ID header attribute. +
@@ -189,6 +242,10 @@ 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" @@ -242,8 +299,8 @@ 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 and correlation id) - . + as well. Optional (by default, the aggregator will use + sequence size) . @@ -254,6 +311,23 @@ present). + + 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 + correlation-strategy, that implements + the completion decision algorithm. Optional, with + restrictions (requires correlation-strategy to be + present). + + The timeout for aggregating messages (counted from the arrival of the first message). Optional. @@ -314,6 +388,25 @@ } }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).
@@ -335,6 +428,11 @@ ... } + @CompletionStrategy + public String correlateBy(OrderItem item) { + ... + } + } @@ -349,8 +447,13 @@ used as the completion strategy of an aggregator. If not present of the method, the aggregator will use the SequenceSizeCompletionStrategy. + - + + An annotation indicating that this method shall be + used as the correlation strategy of an aggregator. If not present of + the method, the aggregator will use the + HeaderAttributeCorrelationStrategy based on CORRELATION_ID.