diff --git a/spring-integration-reference/src/aggregator.xml b/spring-integration-reference/src/aggregator.xml index 2f854a6b3a..0558c46562 100644 --- a/spring-integration-reference/src/aggregator.xml +++ b/spring-integration-reference/src/aggregator.xml @@ -23,65 +23,63 @@
Functionality - The Aggregator combines a group of related messages, by storing - and grouping them, until the group is deemed complete. At that point, - the Aggregator will create a single message by processing the whole - group, and will send the result message further. + The Aggregator combines a group of related messages, by storing and + grouping them, until the group is deemed complete. At that point, the + Aggregator will create a single message by processing the whole group, and + will send the result message further. - 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 - 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. + 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 + 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. - 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 based on their CORRELATION_ID message header (i.e. the messages + with the same CORRELATION_ID will be grouped together). - An important concern with respect to the timeout 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 discarded or not. + An important concern with respect to the timeout 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 + discarded or not.
- The Aggregator API + Programming model - The Aggregation API consists of a number of classes: + The Aggregation API consists of a number of classes: - - - The base class AbstractMessageAggregator and its - subclass MethodInvokingMessageAggregator - - + + + The base class AbstractMessageAggregator and its + subclass MethodInvokingMessageAggregator + + - - - The CompletionStrategy interface and its default - implementation SequenceSizeCompletionStrategy - - + + + The CompletionStrategy interface and its default + implementation SequenceSizeCompletionStrategy + + - 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. + 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. - A brief highlight of the base - AbstractMessageAggregator (the responsibility of - implementing the aggregateMessages method is left to the - developer): + A brief highlight of the base AbstractMessageAggregator + (the responsibility of implementing the aggregateMessages method is left + to the developer): - public abstract class AbstractMessageAggregator + public abstract class AbstractMessageAggregator extends AbstractMessageBarrierConsumer { private volatile CompletionStrategy completionStrategy @@ -92,98 +90,96 @@ } - 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 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: + + 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 { + 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.
Configuring an Aggregator with 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: + 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: - <channel id="inputChannel"/> + <channel id="inputChannel"/> <aggregator id="completelyDefinedAggregator" input-channel="inputChannel" @@ -205,95 +201,94 @@ <bean id="completionStrategyBean" class="sample.PojoCompletionStrategy"/> - - - The id of the aggregator is - optional. - + + + The id of the aggregator is + optional. + - - The input channel of the aggregator. - Required. - + + The input channel of the aggregator. + Required. + - - The channel where the aggregator will send the aggregation - results. Optional (not required, because the aggregator - will honor . - + + The channel where the aggregator will send the aggregation + results. Optional (because incoming messages can specify a + reply channel themselves). + - - The channel where the aggregator will send the messages that - timed out (if send-partial-results-on-timeout is - false). Optional. - + + The channel where the aggregator will send the messages that + timed out (if send-partial-results-on-timeout is + false). 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 reference to a bean defined in the application context. The + bean must implement the aggregation logic as described above. + Required. + - - A method defined on the bean referenced by ref, - that implements the message aggregation - algorithm. Optional, with restrictions (see - above). - + + 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 (by default, the aggregator - . - + + 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-method attribute must be defined + as well. Optional (by default, the aggregator + . + - - A method defined on the bean referenced by - completion-strategy, that - implements the completion decision algorithm. Optional, - with restrictions (requires completion-strategy to be - present). - + + A method defined on the bean referenced by + completion-strategy, that implements + the completion decision algorithm. Optional, with + restrictions (requires completion-strategy to be + present). + - - The timeout for aggregating messages (counted from the arrival - of the first message). 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). - + + 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 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 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. - - + + The timeout for sending out messages. + Optional. + + - An implementation of the aggregator bean, for example, looks as - follows: + An implementation of the aggregator bean, for example, looks as + follows: - public class PojoAggregator { + public class PojoAggregator { public Long add(List<Long> results) { long total = 0l; @@ -305,10 +300,10 @@ } - An implementation of the completion strategy bean for the example - above may be as follows: + An implementation of the completion strategy bean for the example + above may be as follows: - public class PojoCompletionStrategy { + public class PojoCompletionStrategy { ... public boolean checkCompleteness(List<Long> numbers) { int sum = 0; @@ -318,11 +313,11 @@ return sum >= maxValue; } }Wherever it makes sense, the completion strategy method and - the aggregator method can be combined in a single bean. + the aggregator method can be combined in a single bean.
- Configuration an Aggregator with Annotations + Configuring an Aggregator with Annotations An aggregator configured using annotations can look like this. @@ -353,7 +348,7 @@ 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 - SequenceSizeCompletionStrategy. + SequenceSizeCompletionStrategy. diff --git a/spring-integration-reference/src/resequencer.xml b/spring-integration-reference/src/resequencer.xml index 3a655f0b81..2409e7de1d 100644 --- a/spring-integration-reference/src/resequencer.xml +++ b/spring-integration-reference/src/resequencer.xml @@ -8,36 +8,34 @@ Introduction Related to the Aggregator, albeit different from a functional - standpoint, is the Resequencer. In this chapter, we will treat them - together because of their similar functionalities. + standpoint, is the Resequencer.
Functionality - 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. + 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. - 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. + 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.
- Configuring a Resequencer with XML + Configuring a Resequencer with XML - Configuring a resequencer requires only including the appropriate - element in XML. + Configuring a resequencer requires only including the appropriate + element in XML. - A sample resequencer configuration is shown below. + A sample resequencer configuration is shown below. - <channel id="inputChannel"/> + <channel id="inputChannel"/> <channel id="outputChannel"/> @@ -52,69 +50,67 @@ tracked-correlation-id-capacity="99" send-timeout="86420000" /> - - - The id of the resequencer is - optional. - + + + The id of the resequencer is + optional. + - - The input channel of the resequencer. - Required. - + + The input channel of the resequencer. + Required. + - - The channel where the resequencer will send the reordered - messages. Optional. - + + The channel where the resequencer will send the reordered + messages. Optional. + - - The channel where the resequencer will send the messages - that timed out (if send-partial-result-on-timeout is - false). Optional. - + + The channel where the resequencer will send the messages that + timed out (if send-partial-result-on-timeout is + false). Optional. + - - Whether to send out ordered sequences as soon as they are - available, or only after the whole message group arrives. - Optional (true by default). - + + Whether to send out ordered sequences as soon as they are + available, or only after the whole message group arrives. + Optional (true by default). + - - The timeout for reordering message sequences (counted from - the arrival of the first message). - Optional. - + + The timeout for reordering message sequences (counted from the + arrival of the first message). Optional. + - - Whether, upon the expiration of the timeout, the ordered - group shall be sent out (even if some of the messages are - missing). Optional (false by default). - + + Whether, upon the expiration of the timeout, the ordered group + shall 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 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 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. - - - - - Since there is no custom behaviour to be implemented in Java classes for - resequencers, there is no annotation support for it. - + + The timeout for sending out messages. + Optional. + + + + + Since there is no custom behaviour to be implemented in Java classes for resequencers, there is no annotation support for it. +
diff --git a/spring-integration-reference/src/splitter.xml b/spring-integration-reference/src/splitter.xml index ec5a356eb5..63cedb96e9 100644 --- a/spring-integration-reference/src/splitter.xml +++ b/spring-integration-reference/src/splitter.xml @@ -5,18 +5,137 @@ Message Splitter
- The @Splitter Annotation - - The @Splitter annotation is also applicable to methods that expect either the - Message type or the message payload type, and the return values of the method - should be a collection of any type. If the returned values are not actual Message - objects, then each of them will be sent as the payload of a message. Those messages will be sent to the output - channel as designated for the endpoint on which the @Splitter is defined. - @Splitter -List<LineItem> extractItems(Order order) { - return order.getItems() -} - + Introduction + + The Splitter is a component whose role is to partition a message in + several parts, and send the resulting messages to be processed + independently. Very often, they are upstream producers in a pipeline that + includes an Aggregator.
- \ No newline at end of file +
+ Functionality + + +
+ +
+ Programming model + + The API for performing splitting consists from one base class, + AbstractMessageSplitter, which is a MessageConsumer implementation, + encapsulating features which are common to splitters, such as filling in + the appropriate message headers CORRELATION_ID, SEQUENCE_SIZE, and + SEQUENCE_NUMBER on the messages that are produced. This allows to track + down the messages and the results of their processing (in a typical + scenario, these headers would be copied over to the messages that are + produced by the various transforming endpoints), and use them, for + example, in a Composed Message Processor scenario. + + An excerpt from AbstractMessageSplitter can be seen below: + + public abstract class AbstractMessageSplitter + extends AbstractReplyProducingMessageConsumer { + ... + protected abstract Object splitMessage(Message<?> message); + +} + + For implementing a specific Splitter in an application, a developer + can extend AbstractMessageSplitter and implement the splitMessage method, + thus defining the actual logic for splitting the messages. The return + value can be one of the following: + + + + a Collection (or subclass thereof) or an array of Message + objects - in this case the messages will be sent as such (after the + CORRELATION_ID, SEQUENCE_SIZE and SEQUENCE_NUMBER will be populated). + Using this approach gives more control to the developer, for example + for populating custom message headers as part of the splitting + process. + + + + a Collection (or subclass thereof) or an array of non-Message + objects - works like the prior case, except that each collection + element will be used as a Message payload. Using this approach allows + to focus on the + + + + a Message or non-Message object (but not a Collection or an + Array) - it works like the previous cases, except that there is a + single message to be sent out. + + + + In Spring Integration, any POJO can implement the splitting + algorithm, provided that it defines a method that accepts a single + argument and has a return value. In this case, the return value of the + method will be interpreted as described above. The input argument might + either be a Message or a simple POJO. In the latter case, the splitter + will receive the payload of the incoming message. +
+ +
+ Configuring a Splitter using XML + + A splitter can be configured through XML as follows:<channel id="inputChannel"/> + +<splitter id="splitter" + ref="splitterBean" + method="split" + input-channel="inputChannel" + output-channel="outputChannel" /> + +<channel id="outputChannel"/> + +<beans:bean id="splitterBean" class="sample.PojoSplitter"/> + + The id of the splitter is + optional. + + + + A reference to a bean defined in the application context. The + bean must implement the splitting logic as described in the section + above. Required. + + + + The method (defined on the bean specified above) that + implements the splitting logic. + Optional. + + + + The input channel of the splitter. + Required. + + + + The channel where the splitter will send the results of + splitting the incoming message. Optional (because incoming + messages can specify a reply channel themselves). + + +
+ +
+ Configuring a Splitter with Annotations + + The @Splitter annotation is + applicable to methods that expect either the + Message type or the message payload type, + and the return values of the method should be a collection of any type. If + the returned values are not actual Message + objects, then each of them will be sent as the payload of a message. Those + messages will be sent to the output channel as designated for the endpoint + on which the @Splitter is defined. + @Splitter +List<LineItem> extractItems(Order order) { + return order.getItems() +} +
+