Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1128: Docs for MessageStoreReaper

Browse Source
This commit is contained in:
David Syer
2010-06-23 16:54:50 +00:00
parent 9d0080d33a
commit 91452275ad
2 changed files with 319 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

441
src/docbkx/aggregator.xml
View File

@@ -14,9 +14,8 @@
<para>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. In order to do this
it requires a MessageStore</para>
decide when the complete group of Messages is available. In order to do
this it requires a MessageStore</para>
</section>
<section id="aggregator-functionality">
@@ -27,20 +26,20 @@
Aggregator will create a single message by processing the whole group, and
will send that aggregated message as output.</para>
<para>An main 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. The other two aspects are correlation and release</para>
<para>An main 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. The other two aspects are correlation and
release</para>
<para>In Spring Integration, the grouping of the messages for aggregation (correlation)
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 other ways of
specifying how the messages should be grouped together, by using a
CorrelationStrategy (see below).</para>
<para>In Spring Integration, the grouping of the messages for aggregation
(correlation) 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
other ways of specifying how the messages should be grouped together, by
using a CorrelationStrategy (see below).</para>
<para>To determine whether or not a group of messages may be processed,
a ReleaseStrategy is consulted. The default release strategy for aggregator
<para>To determine whether or not a group of messages may be processed, a
ReleaseStrategy is consulted. The default release strategy for aggregator
will release groups that have all messages from the sequence, but this can
be entirely customized</para>
</section>
@@ -52,9 +51,10 @@
<itemizedlist>
<listitem>
<para>The interface <code>MessageGroupProcessor</code> and related
base class <code>AbstractAggregatingMessageGroupProcessor</code> and its
subclass <code>MethodInvokingAggregatingMessageGroupProcessor</code></para>
<para>The interface <code>MessageGroupProcessor</code> and related
base class <code>AbstractAggregatingMessageGroupProcessor</code> and
its subclass
<code>MethodInvokingAggregatingMessageGroupProcessor</code></para>
</listitem>
<listitem>
@@ -69,76 +69,106 @@
</itemizedlist>
<section>
<title>CorrelatingMessageHandler</title>
<para>The <code>CorrelatingMessageHandler</code> is a
<code>MessageHandler</code> implementation, encapsulating the common
functionalities of an Aggregator (and other correlating use cases), which are:
<itemizedlist>
<listitem>
<para>correlating messages into a group to be aggregated</para>
</listitem>
<listitem>
<para>maintaining those messages in a MessageStore until the group may be released</para>
</listitem>
<listitem>
<para>deciding when the group is in fact may be released</para>
</listitem>
<listitem>
<para>processing the released group into a single aggregated message</para>
</listitem>
<listitem>
<para>recognizing and responding to an expired group</para>
</listitem>
</itemizedlist>
The responsibility of deciding how the messages should be grouped together
is delegated to a <code>CorrelationStrategy</code> instance. The responsibility
of deciding whether the message group can be released is delegated to a
<code>ReleaseStrategy</code> instance.</para>
functionalities of an Aggregator (and other correlating use cases),
which are: <itemizedlist>
<listitem>
<para>correlating messages into a group to be aggregated</para>
</listitem>
<listitem>
<para>maintaining those messages in a MessageStore until the group
may be released</para>
</listitem>
<listitem>
<para>deciding when the group is in fact may be released</para>
</listitem>
<listitem>
<para>processing the released group into a single aggregated
message</para>
</listitem>
<listitem>
<para>recognizing and responding to an expired group</para>
</listitem>
</itemizedlist> The responsibility of deciding how the messages should
be grouped together is delegated to a <code>CorrelationStrategy</code>
instance. The responsibility of deciding whether the message group can
be released is delegated to a <code>ReleaseStrategy</code>
instance.</para>
<para>Here is a brief highlight of the base
<code>AbstractAggregatingMessageGroupProcessor</code> (the responsibility of
implementing the aggregateMessages method is left to the
developer):</para>
<code>AbstractAggregatingMessageGroupProcessor</code> (the
responsibility of implementing the aggregateMessages method is left to
the developer):</para>
<programlisting language="java">public abstract class AbstractAggregatingMessageGroupProcessor
<programlisting language="java"><![CDATA[public abstract class AbstractAggregatingMessageGroupProcessor
implements MessageGroupProcessor {
protected Map&lt;String, Object&gt; aggregateHeaders(MessageGroup group) {
protected Map<String, Object> aggregateHeaders(MessageGroup group) {
....
}
protected abstract Object aggregatePayloads(MessageGroup group);
}</programlisting>
The CorrelationStrategy is owned by the <code>CorrelatingMessageHandler</code> and it has
a default value based on the correlation ID message header:
<programlisting language="java">private volatile CorrelationStrategy correlationStrategy =
new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);</programlisting>
}]]></programlisting>
<para>When appropriate, the simplest option is the <code>DefaultAggregatingMessageGroupProcessor</code>.
It creates a single Message whose payload is a List of the payloads received
for a given group. It uses the default <code>CorrelationStrategy</code> and
<code>CompletionStrategy</code> as shown above. This works well for simple
Scatter Gather implementations with either a Splitter, Publish Subscribe Channel,
or Recipient List Router upstream.</para>
The CorrelationStrategy is owned by the
<code>CorrelatingMessageHandler</code>
and it has a default value based on the correlation ID message header:
<programlisting language="java"><![CDATA[private volatile CorrelationStrategy correlationStrategy =
new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);]]></programlisting>
<para>When appropriate, the simplest option is the
<code>DefaultAggregatingMessageGroupProcessor</code>. It creates a
single Message whose payload is a List of the payloads received for a
given group. It uses the default <code>CorrelationStrategy</code> and
<code>CompletionStrategy</code> as shown above. This works well for
simple Scatter Gather implementations with either a Splitter, Publish
Subscribe Channel, or Recipient List Router upstream.</para>
<note>
<para>When using a Publish Subscribe Channel or Recipient List Router in this
type of scenario, be sure to enable the flag to <emphasis>apply-sequence</emphasis>.
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.</para>
<para>When using a Publish Subscribe Channel or Recipient List Router
in this type of scenario, be sure to enable the flag to
<emphasis>apply-sequence</emphasis>. 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.</para>
</note>
<para>When implementing a specific aggregator object for an application,
a developer can extend <code>AbstractAggregatingMessageGroupProcessor</code> and
implement the <code>aggregatePayloads</code> 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.</para>
a developer can extend
<code>AbstractAggregatingMessageGroupProcessor</code> and implement the
<code>aggregatePayloads</code> 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.</para>
<para>In general, any ordinary Java class (i.e. POJO) can implement the
aggregation algorithm. For doing so, it must provide a method that
@@ -146,6 +176,8 @@
supported as well). This method will be invoked for aggregating
messages, as follows:</para>
<itemizedlist>
<listitem>
<para>if the argument is a parametrized java.util.List, and the
@@ -167,12 +199,16 @@
</listitem>
</itemizedlist>
<note>
<para>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.</para>
</note>
</section>
<section>
@@ -181,11 +217,11 @@
<para>The <code>ReleaseStrategy</code> interface is defined as
follows:</para>
<programlisting language="java">public interface ReleaseStrategy {
<programlisting language="java"><![CDATA[public interface ReleaseStrategy {
boolean canRelease(MessageGroup messages);
}</programlisting>
}]]></programlisting>
<para>In general, any ordinary Java class (i.e. POJO) can implement the
completion decision mechanism. For doing so, it must provide a method
@@ -208,27 +244,27 @@
</listitem>
<listitem>
<para>the method must return true if the message group is ready
for aggregation, and false otherwise.</para>
<para>the method must return true if the message group is ready for
aggregation, and false otherwise.</para>
</listitem>
</itemizedlist>
<para>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 is removed from the message store.
Partial sequences can be released, in which case the next time
the <code>ReleaseStrategy</code> is called it will be presented
with a group containing marked messages (already processed) and
unmarked messages (a potential new partial sequence)</para>
<para>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
is removed from the message store. Partial sequences can be released, in
which case the next time the <code>ReleaseStrategy</code> is called it
will be presented with a group containing marked messages (already
processed) and unmarked messages (a potential new partial
sequence)</para>
<para>Spring Integration provides an out-of-the box implementation for
<code>ReleaseStrategy</code>, the
<code>SequenceSizerReleaseStrategy</code>. 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. As shown above, it is also the default strategy.</para>
<code>SequenceSizerReleaseStrategy</code>. 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. As shown
above, it is also the default strategy.</para>
</section>
<section>
@@ -237,11 +273,11 @@
<para>The <code>CorrelationStrategy</code> interface is defined as
follows:</para>
<programlisting language="java">public interface CorrelationStrategy {
<programlisting language="java"><![CDATA[public interface CorrelationStrategy {
Object getCorrelationKey(Message&lt;?&gt; message);
Object getCorrelationKey(Message<?> message);
}</programlisting>
}]]></programlisting>
<para>The method shall return an Object which represents the correlation
key used for grouping messages together. The key must satisfy the
@@ -249,8 +285,8 @@
equals() and hashCode().</para>
<para>In general, any ordinary Java class (i.e. POJO) can implement the
correlation decision mechanism, and the rules for mapping a message to
a method's argument (or arguments) are the same as for a
correlation decision mechanism, and the rules for mapping a message to a
method's argument (or arguments) are the same as for a
<code>ServiceActivator</code> (including support for @Header
annotations). The method must return a value, and the value must not be
<code>null</code>.</para>
@@ -269,34 +305,34 @@
<title>Configuring an Aggregator with XML</title>
<para>Spring Integration supports the configuration of an aggregator via
XML through the &lt;aggregator/&gt; element. Below you can see an example of
an aggregator with all optional parameters defined.</para>
XML through the &lt;aggregator/&gt; element. Below you can see an example
of an aggregator with all optional parameters defined.</para>
<programlisting lang="xml">&lt;channel id="inputChannel"/&gt;
<programlisting lang="xml"><![CDATA[<channel id="inputChannel"/>
&lt;aggregator id="completelyDefinedAggregator" <co id="aggxml1" />
input-channel="inputChannel" <co id="aggxml2" />
output-channel="outputChannel" <co id="aggxml3" />
discard-channel="discardChannel" <co id="aggxml4" />
ref="aggregatorBean" <co id="aggxml5" />
method="add" <co id="aggxml6" />
release-strategy="releaseStrategyBean" <co id="aggxml7" />
release-strategy-method="canRelease" <co id="aggxml8" />
correlation-strategy="correlationStrategyBean" <co
id="aggxmlCorrelationStrategy" />
correlation-strategy-method="groupNumbersByLastDigit" <co
id="aggxmlCorrelationStrategyMethod" />
message-store="messageStore" <co id="aggxml11-co" linkends="aggxml11"/>
send-partial-result-on-expiry="true" <co id="aggxml9" />
send-timeout="86420000" <co id="aggxml10" /> /&gt;
<aggregator id="completelyDefinedAggregator" ]]><co id="aggxml1" /><![CDATA[
input-channel="inputChannel" ]]><co id="aggxml2" /><![CDATA[
output-channel="outputChannel" ]]><co id="aggxml3" /><![CDATA[
discard-channel="discardChannel" ]]><co id="aggxml4" /><![CDATA[
ref="aggregatorBean" ]]><co id="aggxml5" /><![CDATA[
method="add" ]]><co id="aggxml6" /><![CDATA[
release-strategy="releaseStrategyBean" ]]><co id="aggxml7" /><![CDATA[
release-strategy-method="canRelease" ]]><co id="aggxml8" /><![CDATA[
correlation-strategy="correlationStrategyBean" ]]><co
id="aggxmlCorrelationStrategy" /><![CDATA[
correlation-strategy-method="groupNumbersByLastDigit" ]]><co
id="aggxmlCorrelationStrategyMethod" /><![CDATA[
message-store="messageStore" ]]><co id="aggxml11-co" linkends="aggxml11" /><![CDATA[
send-partial-result-on-expiry="true" ]]><co id="aggxml9" /><![CDATA[
send-timeout="86420000" ]]><co id="aggxml10" /><![CDATA[ />
&lt;channel id="outputChannel"/&gt;
<channel id="outputChannel"/>
&lt;bean id="aggregatorBean" class="sample.PojoAggregator"/&gt;
<bean id="aggregatorBean" class="sample.PojoAggregator"/>
&lt;bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/&gt;
<bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/>
&lt;bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/&gt;</programlisting>
<bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/>]]></programlisting>
<calloutlist>
<callout arearefs="aggxml1">
@@ -368,49 +404,49 @@
present).</emphasis></para>
</callout>
<callout arearefs="aggxml11-co" id="aggxml11">
<para>A reference to a <code>MessageGroupStore</code> that
can be used to store groups of messages under their
correlation key until they are
complete. <emphasis>Optional</emphasis> with default a
volatile in-memory store.</para>
</callout>
<callout arearefs="aggxml11-co" id="aggxml11">
<para>A reference to a <code>MessageGroupStore</code> that can be used
to store groups of messages under their correlation key until they are
complete. <emphasis>Optional</emphasis> with default a volatile
in-memory store.</para>
</callout>
<callout arch="" arearefs="aggxml9">
<para>Whether upon the expiration of the message group, the aggregator will
try to aggregate the messages that have already arrived. <emphasis>Optional
(false by default)</emphasis>.</para>
<para>Whether upon the expiration of the message group, the aggregator
will try to aggregate the messages that have already arrived.
<emphasis>Optional (false by default)</emphasis>.</para>
</callout>
<callout arearefs="aggxml10">
<para>The timeout for sending the aggregated messages to the
output or reply channel. <emphasis>Optional</emphasis>.</para>
<para>The timeout for sending the aggregated messages to the output or
reply channel. <emphasis>Optional</emphasis>.</para>
</callout>
</calloutlist>
<para>Using a "ref" attribute is generally recommended if a custom aggregator handler
implementation can be reused in other <code>&lt;aggregator&gt;</code> definitions.
However if a custom aggregator handler implementation should be scoped to a concrete
definition of the <code>&lt;aggregator&gt;</code>, you can use an inner bean definition
(starting with version 1.0.3) for custom aggregator handlers within the
<code>&lt;aggregator&gt;</code> element:
<programlisting language="xml"><![CDATA[<aggregator input-channel="input" method="sum" output-channel="output">
<para>Using a "ref" attribute is generally recommended if a custom
aggregator handler implementation can be reused in other
<code>&lt;aggregator&gt;</code> definitions. However if a custom
aggregator handler implementation should be scoped to a concrete
definition of the <code>&lt;aggregator&gt;</code>, you can use an inner
bean definition (starting with version 1.0.3) for custom aggregator
handlers within the <code>&lt;aggregator&gt;</code> element:
<programlisting language="xml"><![CDATA[<aggregator input-channel="input" method="sum" output-channel="output">
<beans:bean class="org.foo.ExampleAggregator"/>
</aggregator>]]></programlisting>
</para>
</aggregator>]]></programlisting></para>
<note>
<para>Using both a "ref" attribute and an inner bean definition in the same
<code>&lt;aggregator&gt;</code> configuration is not allowed, as it creates an
ambiguous condition. In such cases, an Exception will be thrown.
</para>
<para>Using both a "ref" attribute and an inner bean definition in the
same <code>&lt;aggregator&gt;</code> configuration is not allowed, as it
creates an ambiguous condition. In such cases, an Exception will be
thrown.</para>
</note>
<para>An example implementation of the aggregator bean looks as follows:</para>
<para>An example implementation of the aggregator bean looks as
follows:</para>
<programlisting language="java">public class PojoAggregator {
<programlisting language="java"><![CDATA[public class PojoAggregator {
public Long add(List&lt;Long&gt; results) {
public Long add(List<Long> results) {
long total = 0l;
for (long partialResult: results) {
total += partialResult;
@@ -418,37 +454,34 @@
return total;
}
}</programlisting>
}]]></programlisting>
<para>An implementation of the completion strategy bean for the example
above may be as follows:</para>
<para><programlisting language="java">public class PojoReleaseStrategy {
<para><programlisting language="java"><![CDATA[public class PojoReleaseStrategy {
...
public boolean canRelease(List&lt;Long&gt; numbers) {
public boolean canRelease(List<Long> numbers) {
int sum = 0;
for (long number: numbers) {
sum += number;
}
return sum &gt;= maxValue;
return sum >= maxValue;
}
}</programlisting>
<note>
<para>Wherever it makes sense, the release strategy method and
the aggregator method can be combined in a single bean.</para>
</note>
</para>
}]]></programlisting> <note>
<para>Wherever it makes sense, the release strategy method and the
aggregator method can be combined in a single bean.</para>
</note></para>
<para>An implementation of the correlation strategy bean for the example
above may be as follows:</para>
<para><programlisting language="java">public class PojoCorrelationStrategy {
<para><programlisting language="java"><![CDATA[public class PojoCorrelationStrategy {
...
public Long groupNumbersByLastDigit(Long number) {
return number % 10;
}
}</programlisting></para>
}]]></programlisting></para>
<para>For example, this aggregator would group numbers by some criterion
(in our case the remainder after dividing by 10) and will hold the group
@@ -457,36 +490,109 @@
<note>
<para>Wherever it makes sense, the release strategy method, correlation
strategy method and the aggregator method can be combined in a single bean
(all of them or any two).</para>
strategy method and the aggregator method can be combined in a single
bean (all of them or any two).</para>
</note>
</section>
<section>
<title id="reaper">Managing State in an Aggregator:
MessageGroupStore</title>
<para>Aggregator (and some other patterns in Spring Integration) is a
stateful pattern that requires decisions to be made based on a group of
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. <classname>ReleaseStrategy</classname>) is driven by the principle
that the components (framework and user) should be to remain stateless.
All state is carried by the <classname>MessageGroup</classname> and its
management is delegated to the
<classname>MessageGroupStore</classname>.</para>
<para>The <classname>MessageGroupStore</classname> accumulates state
information in <classname>MessageGroups</classname>, potentially forever.
So to prevent stale state from hanging around, and for volatile stores to
provide a hook for cleaning up when the application shots down, the
<classname>MessageGroupStore</classname> allows the user to register
callbacks to apply to <classname>MessageGroups</classname> when they
expire. The interface is very straighforward:</para>
<programlisting><![CDATA[public interface MessageGroupCallback {
void execute(MessageGroupStore messageGroupStore, MessageGroup group);
}]]></programlisting>
<para>The callback has access directly to the store and the message group
so it can manage the persistent state (e.g. by removing the group from the
store entirely).</para>
<para>The MessageGroupStore maintains a list of these callbacks which it
applies when asked to all messages whose timestamp is earlier than a time
supplied as a parameter:</para>
<programlisting><![CDATA[public interface MessageGroupStore {
void registerMessageGroupExpiryCallback(MessageGroupCallback callback);
int expireMessageGroups(long timeout);
}]]></programlisting>
<para>The expireMessageGroups method can be called with a timeout value:
any message older than the current time minus this value wiull be expired,
and have the callbacks applied. Thus it is the user of the store that
defines what is meant by message group "expiry".</para>
<para>As a convenience for users, Spring Integration provides a wrapper
for the message expiry in the form of a
<classname>MessageGroupStoreReaper</classname>:</para>
<programlisting><![CDATA[<bean id="reaper" class="org...MessageGroupStoreReaper">
<property name="messageGroupStore" ref="messageStore"/>
<property name="timeout" value="10"/>
</bean>
<task:scheduled-tasks scheduler="scheduler">
<task:scheduled ref="reaper" method="run" fixed-rate="10000"/>
</task:scheduled-tasks>]]></programlisting>
<para>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
<classname>CorrelatingMessageHandler</classname>.</para>
<para>The <classname>CorrelatingMessageHandler</classname> registers its
own expiry callback, and this is the link with the boolean flag
<code>send-partial-result-on-expiry</code> 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.</para>
</section>
<section id="aggregator-annotations">
<title>Configuring an Aggregator with Annotations</title>
<para>An aggregator configured using annotations can look like
this.</para>
<programlisting language="java">public class Waiter {
<programlisting language="java"><![CDATA[public class Waiter {
...
@Aggregator <co id="aggann" />
public Delivery aggregatingMethod(List&lt;OrderItem&gt; items) {
@Aggregator ]]><co id="aggann" /><![CDATA[
public Delivery aggregatingMethod(List<OrderItem> items) {
...
}
@ReleaseStrategy <co id="agganncs" />
public boolean releaseChecker(List&lt;Message&lt;?&gt;&gt; messages) {
@ReleaseStrategy ]]><co id="agganncs" /><![CDATA[
public boolean releaseChecker(List<Message<?>> messages) {
...
}
@CorrelationStrategy <co id="agganncorrs" />
@CorrelationStrategy ]]><co id="agganncorrs" /><![CDATA[
public String correlateBy(OrderItem item) {
...
}
}</programlisting>
}]]></programlisting>
<calloutlist>
<callout arearefs="aggann">
@@ -497,8 +603,8 @@
<callout arearefs="agganncs">
<para id="aggann2">An annotation indicating that this method shall be
used as the release strategy of an aggregator. If not present on
any method, the aggregator will use the
used as the release strategy of an aggregator. If not present on any
method, the aggregator will use the
SequenceSizeCompletionStrategy.</para>
</callout>
@@ -510,12 +616,11 @@
</callout>
</calloutlist>
<para>All of the configuration options provided by the xml element are also
available for the @Aggregator annotation.</para>
<para>All of the configuration options provided by the xml element are
also available for the @Aggregator annotation.</para>
<para>The aggregator can be either referenced explicitly from XML or, if
the @MessageEndpoint is defined on the class, detected automatically
through classpath scanning.</para>
</section>
</chapter>

74
src/docbkx/resequencer.xml
View File

@@ -8,7 +8,7 @@
<title>Introduction</title>
<para>Related to the Aggregator, albeit different from a functional
standpoint, is the Resequencer. </para>
standpoint, is the Resequencer.</para>
</section>
<section id="resequencer-functionality">
@@ -16,9 +16,9 @@
<para>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 the Resequencer does not process the messages in
any way. It simply releases them in the order of their SEQUENCE_NUMBER
header values.</para>
difference being that the Resequencer does not process the messages in any
way. It simply releases them in the order of their SEQUENCE_NUMBER header
values.</para>
<para>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
@@ -33,18 +33,20 @@
<para>A sample resequencer configuration is shown below.</para>
<programlisting language="xml">&lt;channel id="inputChannel"/&gt;
<programlisting language="xml"><![CDATA[<channel id="inputChannel"/>
&lt;channel id="outputChannel"/&gt;
<channel id="outputChannel"/>
&lt;resequencer id="completelyDefinedResequencer" <co id="resxml1-co" linkends="resxml1" />
input-channel="inputChannel" <co id="resxml2-co" linkends="resxml2"/>
output-channel="outputChannel" <co id="resxml3-co" linkends="resxml3"/>
discard-channel="discardChannel" <co id="resxml4-co" linkends="resxml4"/>
release-partial-sequences="true" <co id="resxml5-co" linkends="resxml5"/>
message-store="messageStore" <co id="resxml6-co" linkends="resxml6"/>
send-partial-result-on-expiry="true" <co id="resxml7-co" linkends="resxml7"/>
send-timeout="86420000" <co id="resxml10-co" linkends="resxml10"/> /&gt; </programlisting>
<resequencer id="completelyDefinedResequencer" ]]><co id="resxml1-co"
linkends="resxml1" /><![CDATA[
input-channel="inputChannel" ]]><co id="resxml2-co" linkends="resxml2" /><![CDATA[
output-channel="outputChannel" ]]><co id="resxml3-co" linkends="resxml3" /><![CDATA[
discard-channel="discardChannel" ]]><co id="resxml4-co" linkends="resxml4" /><![CDATA[
release-partial-sequences="true" ]]><co id="resxml5-co" linkends="resxml5" /><![CDATA[
message-store="messageStore" ]]><co id="resxml6-co" linkends="resxml6" /><![CDATA[
send-partial-result-on-expiry="true" ]]><co id="resxml7-co"
linkends="resxml7" /><![CDATA[
send-timeout="86420000" ]]><co id="resxml10-co" linkends="resxml10" /><![CDATA[ /> ]]></programlisting>
<para><calloutlist>
<callout arearefs="resxml1-co" id="resxml1">
@@ -69,31 +71,47 @@
</callout>
<callout arearefs="resxml5-co" id="resxml5">
<para>Whether to send out ordered sequences as soon as they are
available, or only after the whole message group arrives.
<emphasis>Optional (false by default)</emphasis>.</para> If this
flag is not specified (so a complete sequence is defined by the sequence
headers) then it can make sense to provide a custom <code>Comparator</code>
to be used to order the messages when sending
(use the XML attribute <literal>comparator</literal> to point to a bean
definition). If <literal>release-partial-sequences</literal> is true then
there is no way with a custom comparator to define a partial sequence. To
do that you would have to provide a <literal>release-strategy</literal> (also
a reference to another bean definition, either a POJO or a <code>ReleaseStrategy</code>).
<emphasis>Optional (false by default)</emphasis>.</para>
If this flag is not specified (so a complete sequence is defined by the sequence headers) then it can make sense to provide a custom
<code>Comparator</code>
to be used to order the messages when sending (use the XML attribute
<literal>comparator</literal>
to point to a bean definition). If
<literal>release-partial-sequences</literal>
is true then there is no way with a custom comparator to define a partial sequence. To do that you would have to provide a
<literal>release-strategy</literal>
(also a reference to another bean definition, either a POJO or a
<code>ReleaseStrategy</code>
).
</callout>
<callout arearefs="resxml6-co" id="resxml6">
<para>A reference to a <code>MessageGroupStore</code> that
can be used to store groups of messages under their
correlation key until they are
complete. <emphasis>Optional</emphasis> with default a
<para>A reference to a <code>MessageGroupStore</code> that can be
used to store groups of messages under their correlation key until
they are complete. <emphasis>Optional</emphasis> with default a
volatile in-memory store.</para>
</callout>
<callout arearefs="resxml7-co" id="resxml7">
<para>Whether, upon the expiration of the group, the ordered group
should be sent out (even if some of the messages are missing).
<emphasis>Optional (false by default)</emphasis>.</para>
<emphasis>Optional (false by default)</emphasis>. See <xref
linkend="reaper" />.</para>
</callout>
<callout arearefs="resxml10-co" id="resxml10">