diff --git a/spring-integration-reference/src/aggregator.xml b/spring-integration-reference/src/aggregator.xml index 907da1f2c2..764039bed4 100644 --- a/spring-integration-reference/src/aggregator.xml +++ b/spring-integration-reference/src/aggregator.xml @@ -78,15 +78,30 @@ 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 + functionalities of an Aggregator, which are: + + + correlating messages into a group to be aggregated + + + maintaining those messages until the group is complete + + + deciding when the group is in fact complete + + + processing the completed group into a single aggregated message + + + recognizing and responding to a timed-out completion attempt + + + 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 + of deciding whether the message group is complete is delegated to a CompletionStrategy instance. - A brief highlight of the base + Here is a brief highlight of the base AbstractMessageAggregator (the responsibility of implementing the aggregateMessages method is left to the developer): @@ -101,8 +116,28 @@ protected abstract Message<?> aggregateMessages(List<Message<?>> messages); } + It also inherits the following default CorrelationStrategy: + private volatile CorrelationStrategy correlationStrategy = + new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID); - For implementing a specific aggregator object for an application, + When appropriate, the simplest option is the DefaultMessageAggregator. + 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 + Scatter Gather implementations with either a Splitter, Publish Subscribe Channel, + or Recipient List Router upstream. + + + When using a Publish Subscribe Channel or Recipient List Router in this + type of scenario, be sure to enable the flag to apply sequence. + That will add the necessary headers (correlation id, sequence number and sequence + size). That behavior is enabled by default for Splitters in Spring Integration, + but it is not enabled for the Publish Subscribe Channel or Recipient List + Router because those components may be used in a variety of contexts where + those headers are not necessary. + + + When 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 @@ -187,7 +222,7 @@ 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. + aggregated. As shown above, it is also the default strategy.
@@ -330,13 +365,14 @@ - The timeout for aggregating messages (counted from the arrival - of the first message). Optional. + The timeout (in milliseconds) 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 + try to aggregate the messages that have already arrived. Optional (false by default). @@ -350,31 +386,36 @@ 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). + 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 out messages. - Optional. + The timeout for sending the aggregated messages to the + output or reply channel. Optional. - - Using a "ref" attribute is generally recommended if custom aggregator handler implementation can be reused in other <aggregator> definitions. However -if custom aggregator handler implementation has to be scoped to a concrete definition of the <aggregator>, starting with v1.0.3, Spring Integration supports -inner bean definitions for custom aggregator handlers within the <aggregator> element: - - + + Using a "ref" attribute is generally recommended if a custom aggregator handler + implementation can be reused in other <aggregator> definitions. + However if a custom aggregator handler implementation should be scoped to a concrete + definition of the <aggregator>, you can use an inner bean definition + (starting with version 1.0.3) for custom aggregator handlers within the + <aggregator> element: + + ]]> + - - Using both "ref" attribute and inner handler definition in the same <aggregator> configuration - is not allowed, as it creates an ambiguous condition and will result in Exception being thrown - + Using both a "ref" attribute and an inner bean definition in the same + <aggregator> configuration is not allowed, as it creates an + ambiguous condition. In such cases, an Exception will be thrown. + - An implementation of the aggregator bean, for example, looks as - follows: + + An example implementation of the aggregator bean looks as follows: public class PojoAggregator { @@ -400,8 +441,13 @@ inner bean definitions for custom aggregator handlers within the <aggre } return sum >= maxValue; } -}Wherever it makes sense, the completion strategy method and - the aggregator method can be combined in a single bean. +} + + + 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: @@ -418,9 +464,11 @@ inner bean definitions for custom aggregator handlers within the <aggre 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). + + 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). +
@@ -458,8 +506,8 @@ inner bean definitions for custom aggregator handlers within the <aggre An annotation indicating that this method shall be - used as the completion strategy of an aggregator. If not present of - the method, the aggregator will use the + used as the completion strategy of an aggregator. If not present on + any method, the aggregator will use the SequenceSizeCompletionStrategy. @@ -471,11 +519,12 @@ inner bean definitions for custom aggregator handlers within the <aggre - All the configuration options provided by the xml element are also + All of the configuration options provided by the xml element are also available for the @Aggregator annotation. The aggregator can be either referenced explicitly from XML or, if the @MessageEndpoint is defined on the class, detected automatically through classpath scanning. +