From 0f3a37ee472b1837550ed9215cd4ee3a2009bb4b Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Fri, 17 Oct 2008 04:44:29 +0000 Subject: [PATCH] Initial aggregation and resequencing content - IN PROGRESS --- .../src/aggregator-resequencer.xml | 166 ++++++++++++++++++ .../src/spring-integration-reference.xml | 1 + 2 files changed, 167 insertions(+) create mode 100644 spring-integration-reference/src/aggregator-resequencer.xml diff --git a/spring-integration-reference/src/aggregator-resequencer.xml b/spring-integration-reference/src/aggregator-resequencer.xml new file mode 100644 index 0000000000..13f91a977d --- /dev/null +++ b/spring-integration-reference/src/aggregator-resequencer.xml @@ -0,0 +1,166 @@ + + + + Aggregation and Resequencing + +
+ Introduction + + Basically a mirror-image of the Splitter, the Aggregator is a type + of Message Consumer 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. + + 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. + + 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. +
+ +
+ Functionality + +
+ Aggregation + + 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. + + 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. +
+ +
+ Resequencing + + 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. +
+
+ +
+ The API + + + +
+ Programming with Aggregator + + The Aggregation API consists of a number of classes: + + + + The base class AbstractMessageAggregator and its + subclass MethodInvokingMessageAggregator + + + + + + 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. + + A brief highlight of the base + AbstractMessageAggregator (the responsibility of + implementing the aggregateMessages method is left to the + developer): + + public abstract class AbstractMessageAggregator extends AbstractMessageBarrierConsumer { + + private volatile CompletionStrategy completionStrategy + = new SequenceSizeCompletionStrategy(); + .... + + protected abstract Message<?> aggregateMessages(List<Message<?>> messages); + +} + + 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 the interest of code simplicity, and promoting good + practices such as low coupling, testability, etc., the preferred way + of implementing the aggregation logic 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: + + public interface CompletionStrategy { + + boolean isComplete(List<Message<?>> messages); + +} + + 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. +
+
+ +
+ Configuration using XML + + +
+ +
+ Configuration using annotations + + +
+ diff --git a/spring-integration-reference/src/spring-integration-reference.xml b/spring-integration-reference/src/spring-integration-reference.xml index c06b2e4ee0..9de9eef7ea 100644 --- a/spring-integration-reference/src/spring-integration-reference.xml +++ b/spring-integration-reference/src/spring-integration-reference.xml @@ -41,6 +41,7 @@ +