diff --git a/docs/src/reference/docbook/aggregator.xml b/docs/src/reference/docbook/aggregator.xml index 6d7d3b7ad7..3199881200 100644 --- a/docs/src/reference/docbook/aggregator.xml +++ b/docs/src/reference/docbook/aggregator.xml @@ -28,7 +28,7 @@ Implementing an Aggregator requires providing the logic to perform the aggregation (i.e., the creation of a single message from many). Two related concepts are correlation and - release + release. Correlation determines how messages are grouped for aggregation. In Spring Integration correlation is done by default based on the CORRELATION_ID message @@ -37,10 +37,12 @@ other ways of specifying how the messages should be grouped together by implementing a CorrelationStrategy (see below). - To determine the state in which a group of messages may be processed, a + To determine the point at which a group of messages is ready to be processed, a ReleaseStrategy is consulted. - The default release strategy for Aggregator will release groups when all - messages included in the sequence are present but this may be customized. + The default release strategy for the Aggregator will release a group when all + messages included in a sequence are present, based on the SEQUENCE_SIZE header. + This default strategy may be overridden by providing a reference to a + custom ReleaseStrategy implementation.
@@ -50,10 +52,10 @@ - The interface MessageGroupProcessor and related - base class AbstractAggregatingMessageGroupProcessor and - its subclass - MethodInvokingAggregatingMessageGroupProcessor + The interface MessageGroupProcessor, and + its subclasses: + MethodInvokingAggregatingMessageGroupProcessor and + ExpressionEvaluatingMessageGroupProcessor @@ -68,12 +70,8 @@
- - CorrelatingMessageHandler - - The CorrelatingMessageHandler is a MessageHandler implementation, encapsulating the common functionalities of an Aggregator (and other correlating use cases), @@ -104,23 +102,19 @@ be released is delegated to a ReleaseStrategy instance. - - Here is a brief highlight of the base AbstractAggregatingMessageGroupProcessor (the - responsibility of implementing the aggregateMessages method is left to + responsibility of implementing the aggregatePayloads method is left to the developer): - - aggregateHeaders(MessageGroup group) { - .... + // default implementation exists } - protected abstract Object aggregatePayloads(MessageGroup group); + protected abstract Object aggregatePayloads(MessageGroup group, Map defaultHeaders); }]]> @@ -141,9 +135,7 @@ } ]]> - - - When appropriate, the simplest option is the + As for actual processing of the message group, the default implementation is the DefaultAggregatingMessageGroupProcessor. It creates a single Message whose payload is a List of the payloads received for a given group. This works well for simple Scatter Gather implementations with either a Splitter, Publish @@ -160,9 +152,7 @@ contexts in which these headers are not necessary. - - - When implementing a specific aggregator object for an application, + When implementing a specific aggregator strategy for an application, a developer can extend AbstractAggregatingMessageGroupProcessor and implement the aggregatePayloads method. However, there are better solutions, less @@ -170,9 +160,9 @@ either through XML or through annotations. In general, any POJO can implement the - aggregation algorithm if it provide a method that + aggregation algorithm if it provides a method that accepts a single java.util.List as an argument - (parametrized lists are supported as well). This method will be invoked for aggregating + (parameterized lists are supported as well). This method will be invoked for aggregating messages as follows: @@ -183,7 +173,7 @@ - if the argument is a non-parametrized java.util.List or the + if the argument is a non-parameterized java.util.List or the parameter type is not assignable to Message, then the method will receive the payloads of the accumulated messages @@ -195,15 +185,12 @@ - - 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 configuring it in the application. -
@@ -215,15 +202,15 @@ In general, any POJO can implement the - completion decision logic if provide a method - that a single java.util.List as an argument (parametrized lists + completion decision logic if it provides a method that accepts a single + java.util.List as an argument (parameterized 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 + invoked after the arrival of each new message, to decide whether the group is complete or not, as follows: @@ -248,17 +235,17 @@ 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 + 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) + processed) and unmarked messages (potentially a new partial + sequence). Spring Integration provides an out-of-the box implementation for ReleaseStrategy, the - SequenceSizeReleaseStrategy. This implementation uses the - SEQUENCE_NUMBER and SEQUENCE_SIZE of the arriving messages for deciding + SequenceSizeReleaseStrategy. This implementation consults the + SEQUENCE_NUMBER and SEQUENCE_SIZE headers of each arriving message to decide when a message group is complete and ready to be aggregated. As shown above, it is also the default strategy.
@@ -275,8 +262,8 @@ }]]> - The method returns an Object which represents the correlation - key used for grouping messages together. The key must satisfy the + The method returns an Object which represents the correlation key + used for associating the message with a message group. The key must satisfy the criteria used for a key in a Map with respect to the implementation of equals() and hashCode(). @@ -293,7 +280,10 @@ 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. + the value of the CORRELATION_ID header attribute. If you have a custom header name + you would like to use for correlation, then simply configure that on an instance of + HeaderAttributeCorrelationStrategy and provide that as a + reference for the Aggregator's correlation-strategy. @@ -334,7 +324,7 @@ - + @@ -368,7 +358,7 @@ The channel to which the aggregator will send the messages that - timed out (if send-partial-results-on-expiry is + timed out (if send-partial-result-on-expiry is false). Optional. @@ -380,7 +370,7 @@ - Order of this aggregator in when more then one aggregator is subscribing to the same DirectChannel + Order of this aggregator when more than one handle is subscribed to the same DirectChannel (use for load balancing purposes). Optional. @@ -401,7 +391,7 @@ algorithm. 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 - sequence size) . + the CORRELATION_ID header) . @@ -414,7 +404,7 @@ A reference to a bean defined in the application context. The bean must implement the aggregation logic - as described above.Optional (by default the list of aggregated Messages will become a + as described above. Optional (by default the list of aggregated Messages will become a payload of the output message). @@ -428,8 +418,7 @@ The bean can be an implementation of the ReleaseStrategy interface or a POJO. In the latter case the release-strategy-method attribute must be defined as well. Optional (by default, the - aggregator will use the correlation id header attribute) - . + aggregator will use the SEQUENCE_SIZE header attribute). @@ -443,14 +432,14 @@ Using a ref attribute is generally recommended if a custom - aggregator handler implementation my be referenced in other + aggregator handler implementation may be referenced in other <aggregator> definitions. However if a custom - aggregator implementation should be scoped to a single - definition of the <aggregator>, use an inner - bean definition (starting with version 1.0.3) for custom aggregator - handlers within the <aggregator> element: + aggregator implementation is only being used by a single + definition of the <aggregator>, you can use an inner + bean definition (starting with version 1.0.3) to configure the aggregation + POJO within the <aggregator> element: - + ]]> @@ -504,7 +493,7 @@ For example, this aggregator would group numbers by some criterion (in our case the remainder after dividing by 10) and will hold the group - until the sum of the numbers which represents the payload exceeds a + until the sum of the numbers provided by the payloads exceeds a certain value. @@ -518,28 +507,29 @@ - Since Spring Integration 2.0, the release strategy may be handled with + Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with SpEL which is recommended if the logic behind such release strategy is relatively simple. - Let's say you have a legacy component which was designed to receive an array of objects. We know that default release - strategy will assemble all aggregated messages in the List. So now we have two problems. First we need to extract individual - messages form such list, extract payload of each message and assemble them into the array of objects (see code below) - - > mesages){ - List strList = new ArrayList(); - for (Message message : mesages) { - strList.add(message.getPayload()); + Let's say you have a legacy component that was designed to receive an array of objects. We know that the default release + strategy will assemble all aggregated messages in the List. So now we have two problems. First we need to extract + individual messages from the list, and then we need to extract the payload of each message and assemble + the array of objects (see code below). + + > messages){ + List stringList = new ArrayList(); + for (Message message : messages) { + stringList.add(message.getPayload()); } - return strList.toArray(new String[]{}); + return stringList.toArray(new String[]{}); }]]> - However, with SpEL such requirement could actually be handled relatively easy with a simple + However, with SpEL such a requirement could actually be handled relatively easily with a one-line expression, thus sparing you from writing a custom class and configuring it as a bean. - ]]> - + In the above configuration we are using a Collection Projection expression to assemble a new collection from the payloads of all messages in the list and then transforming it to an Array, thus @@ -547,41 +537,41 @@ - The same expression-based approach could be applied when dealing with custom Release and + The same expression-based approach can be applied when dealing with custom Release and Correlation strategies. - Instead of defining a bean for custom CorrelationStrategy via correlation-strategy - attribute you can implement your simple correlation logic via SpEL expression and configure it via - correlation-strategy-expression attribute. + Instead of defining a bean for a custom CorrelationStrategy via + the correlation-strategy attribute, you can implement your simple correlation logic + via a SpEL expression and configure it via the correlation-strategy-expression attribute. For example: - - + + In the above example it is assumed that the payload has an attribute person with an id which is going to be used to correlate messages. - And when it comes to ReleaseStrategy you can implement your release logic via - SpEL expression as well and configure it via release-strategy-expression attribute. + Likewise, for the ReleaseStrategy you can implement your release logic via + a SpEL expression and configure it via the release-strategy-expression attribute. For example: - 5"]]> + - In this example the root of SpEL Evaluation Context is MessageGroup and you simply stating - that as soon as there are more then 5 messages in this group release the group. + In this example the root object of the SpEL Evaluation Context is the + MessageGroup itself, and you are simply stating + that as soon as there are more than 5 messages in this group, it should be released. - +
Configuring an Aggregator with Annotations - An aggregator configured using annotations can look like - this. + An aggregator configured using annotations would look like this. - -
Managing State in an Aggregator: MessageGroupStore @@ -646,18 +634,19 @@ messages that have arrived over a period of time, all with the same correlation key. The design of the interfaces in the stateful patterns (e.g. ReleaseStrategy) is driven by the principle - that the components (framework and user) should be to remain stateless. + that the components (whether defined by the framework or a user) should be able to remain stateless. All state is carried by the MessageGroup and its management is delegated to the MessageGroupStore. The MessageGroupStore accumulates state - information in MessageGroups, potentially forever. + information in MessageGroups while waiting for + a release strategy to be triggered, and that event might not ever happen. So to prevent stale messages from lingering, and for volatile stores to provide a hook for cleaning up when the application shuts down, the MessageGroupStore allows the user to register callbacks to apply to its MessageGroups when they - expire. The interface is very straighforward: + expire. The interface is very straightforward: - The callback has access directly to the store and the message group + The callback has direct access to the store and the message group so it can manage the persistent state (e.g. by removing the group from the store entirely). The MessageGroupStore maintains a list of these callbacks which it - applies when asked to all messages whose timestamp is earlier than a time + applies, on demand, to all messages whose timestamp is earlier than a time supplied as a parameter: - + ]]> - The reaper is a Runnable, and all that is happening is that the - message group store's expire method is being called in the sample above - once every 10 seconds. In addition to the reaper, the expiry callbacks are - invoked when the application shuts down via a lifecycle callback in the - CorrelatingMessageHandler. + The reaper is a Runnable, and all that is happening + in the sample above is that the message group store's expire method is being called + once every 10 seconds. The timeout itself is 30 seconds. + + In addition to the reaper, the expiry callbacks are invoked when the application + shuts down via a lifecycle callback in the CorrelatingMessageHandler. + The CorrelatingMessageHandler registers its own expiry callback, and this is the link with the boolean flag send-partial-result-on-expiry in the XML configuration of the aggregator. If the flag is set to true, then when the expiry callback is - invoked then any unmarked messages in groups that are not yet released can - be sent on to the downstream channel. + invoked, any unmarked messages in groups that are not yet released can + be sent on to the output channel.