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

Merge pull request #683 from garyrussell/INT-2832-2833

Browse Source 
* INT-2832-2833:
  INT-2832/2833 Aggregator Fix and Documentation
This commit is contained in:
Mark Fisher
2012-11-30 15:38:48 -05:00
parent 84f4bea5e7 3ea05576b3
commit 120410f78c
3 changed files with 193 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -92,6 +92,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
private boolean lockRegistrySet = false;
private volatile long minimumTimeoutForEmptyGroups;
public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
Assert.notNull(processor);
@@ -172,6 +174,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
this.sendPartialResultOnExpiry = sendPartialResultOnExpiry;
}
/**
* By default, when a MessageGroupStoreReaper is configured to expire partial
* groups, empty groups are also removed. Empty groups exist after a group
* is released normally. This is to enable the detection and discarding of
* late-arriving messages. If you wish to run empty group deletion on a longer
* schedule than expiring partial groups, set this property. Empty groups will
* then not be removed from the MessageStore until they have not been modified
* for at least this number of milliseconds.
*
* @param minimumTimeoutForEmptyGroups The minimum timeout.
*/
public void setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups) {
this.minimumTimeoutForEmptyGroups = minimumTimeoutForEmptyGroups;
}
public void setReleasePartialSequences(boolean releasePartialSequences){
Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy,
"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName()
@@ -241,7 +258,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
*/
protected abstract void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages);
private final boolean forceComplete(MessageGroup group) {
private void forceComplete(MessageGroup group) {
Object correlationKey = group.getGroupId();
// UUIDConverter is no-op if already converted
@@ -250,41 +267,47 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
try {
lock.lockInterruptibly();
try {
if (group.size() > 0) {
try {
/*
* Need to verify the group hasn't changed while we were waiting on
* its lock. We have to re-fetch the group for this. A possible
* future improvement would be to add MessageGroupStore.getLastModified(groupId).
*/
MessageGroup messageGroupNow = this.messageStore.getMessageGroup(
group.getGroupId());
long lastModifiedNow = messageGroupNow.getLastModified();
if (group.getLastModified() == lastModifiedNow) {
if (releaseStrategy.canRelease(group)) {
this.completeGroup(correlationKey, group);
}
else {
this.expireGroup(correlationKey, group);
}
/*
* Need to verify the group hasn't changed while we were waiting on
* its lock. We have to re-fetch the group for this. A possible
* future improvement would be to add MessageGroupStore.getLastModified(groupId).
*/
MessageGroup messageGroupNow = this.messageStore.getMessageGroup(
group.getGroupId());
long lastModifiedNow = messageGroupNow.getLastModified();
if (group.getLastModified() == lastModifiedNow) {
if (group.size() > 0) {
if (releaseStrategy.canRelease(group)) {
this.completeGroup(correlationKey, group);
}
else {
removeGroup = false;
if (logger.isDebugEnabled()) {
logger.debug("Group expiry candidate (" + group.getGroupId() +
") has changed - it may be reconsidered for a future expiration");
}
this.expireGroup(correlationKey, group);
}
}
finally {
if (removeGroup) {
this.remove(group);
else {
/*
* By default empty groups are removed on the same schedule as non-empty
* groups. A longer timeout for empty groups can be enabled by
* setting minimumTimeoutForEmptyGroups.
*/
removeGroup = lastModifiedNow < (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
if (removeGroup && logger.isDebugEnabled()) {
logger.debug("Removing empty group: " + group.getGroupId());
}
}
return true;
}
else {
removeGroup = false;
if (logger.isDebugEnabled()) {
logger.debug("Group expiry candidate (" + group.getGroupId() +
") has changed - it may be reconsidered for a future expiration");
}
}
}
finally {
if (removeGroup) {
this.remove(group);
}
lock.unlock();
}
}
@@ -292,7 +315,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
Thread.currentThread().interrupt();
throw new MessagingException("Thread was interrupted while trying to obtain lock");
}
return false;
}
void remove(MessageGroup group) {

99
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
View File

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -34,6 +35,7 @@ import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
/**
* @author Gary Russell
@@ -150,4 +152,101 @@ public class AbstractCorrelatingMessageHandlerTests {
assertNull(discards.receive(0));
}
@Test // INT-2833
public void testReaperReapsAnEmptyGroup() throws Exception {
final MessageGroupStore groupStore = new SimpleMessageStore();
AggregatingMessageHandler handler = new AggregatingMessageHandler(
new MessageGroupProcessor() {
public Object processMessageGroup(MessageGroup group) {
return group;
}
}, groupStore) {
};
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
/*
* Executes when group 'bar' completes normally
*/
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
return true;
}
public boolean send(Message<?> message) {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
public boolean canRelease(MessageGroup group) {
return group.size() == 1;
}
});
Message<String> message = MessageBuilder.withPayload("foo")
.setCorrelationId("bar")
.build();
handler.handleMessage(message);
assertEquals(1, outputMessages.size());
assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
groupStore.expireMessageGroups(0);
assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
}
@Test // INT-2833
public void testReaperReapsAnEmptyGroupAfterConfiguredDelay() throws Exception {
final MessageGroupStore groupStore = new SimpleMessageStore();
AggregatingMessageHandler handler = new AggregatingMessageHandler(
new MessageGroupProcessor() {
public Object processMessageGroup(MessageGroup group) {
return group;
}
}, groupStore) {
};
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
/*
* Executes when group 'bar' completes normally
*/
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
return true;
}
public boolean send(Message<?> message) {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
public boolean canRelease(MessageGroup group) {
return group.size() == 1;
}
});
handler.setMinimumTimeoutForEmptyGroups(1000);
Message<String> message = MessageBuilder.withPayload("foo")
.setCorrelationId("bar")
.build();
handler.handleMessage(message);
assertEquals(1, outputMessages.size());
assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
groupStore.expireMessageGroups(0);
assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
Thread.sleep(1010);
groupStore.expireMessageGroups(0);
assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
}
}

65
src/reference/docbook/aggregator.xml
View File

@@ -31,8 +31,9 @@
release.</para>
<para>Correlation determines how messages are grouped for aggregation.
In Spring Integration correlation is done by default based on the CORRELATION_ID message
header. Messages with the same CORRELATION_ID will be grouped
In Spring Integration correlation is done by default based on the
<code>MessageHeaders.CORRELATION_ID</code> message
header. Messages with the same <code>MessageHeaders.CORRELATION_ID</code> will be grouped
together. However, the correlation strategy may be customized to allow
other ways of specifying how the messages should be grouped together by
implementing a <interfacename>CorrelationStrategy</interfacename> (see below).</para>
@@ -40,7 +41,8 @@
<para>To determine the point at which a group of messages is ready to be processed, a
<interfacename>ReleaseStrategy</interfacename> is consulted.
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.
messages included in a sequence are present, based on the
<code>MessageHeaders.SEQUENCE_SIZE</code> header.
This default strategy may be overridden by providing a reference to a
custom <interfacename>ReleaseStrategy</interfacename> implementation.</para>
</section>
@@ -70,9 +72,10 @@
</itemizedlist>
<section>
<title>CorrelatingMessageHandler</title>
<title>AggregatingMessageHandler</title>
<para>The <classname>CorrelatingMessageHandler</classname> is a
<para>The <classname>AggregatingMessageHandler</classname> (subclass of
<classname>AbstractCorrelatingMessageHandler</classname>) is a
<interfacename>MessageHandler</interfacename> implementation, encapsulating the common
functionalities of an Aggregator (and other correlating use cases),
which are: <itemizedlist>
@@ -120,9 +123,9 @@
The <interfacename>CorrelationStrategy</interfacename> is owned by the
<classname>CorrelatingMessageHandler</classname>
<classname>AbstractCorrelatingMessageHandler</classname>
and it has a default value based on the CORRELATION_ID message header:
and it has a default value based on the <code>MessageHeaders.CORRELATION_ID</code> message header:
<programlisting language="java"><![CDATA[
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
@@ -247,22 +250,32 @@
}]]></programlisting>
<para>
As you can see based on the above signatures, the POJO-based Release Strategy will be passed a <classname>Collection</classname> of unmarked Messages
if you need access to the whole <classname>Message</classname> or <classname>Collection</classname> of payload objects if the type parameter is
anything other than <classname>Message</classname>. Typically this would satisfy the majority of use cases. However if
for some reason you need to access the full <classname>MessageGroup</classname> - which contains <code>unmarked</code> and <code>marked</code> Messages -
As you can see based on the above signatures, the POJO-based Release Strategy will be passed a
<classname>Collection</classname> of not-yet-released Messages
(if you need access to the whole <classname>Message</classname>) or a <classname>Collection</classname> of payload objects
(if the type parameter is
anything other than <classname>Message</classname>). Typically this would satisfy the majority of use cases. However if,
for some reason, you need to access the full <classname>MessageGroup</classname>
then you should simply provide an implementation of the <classname>ReleaseStrategy</classname> interface.
</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
<para>When the group is released for aggregation, all its not-yet-released
messages are processed and removed from the group.
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 <interfacename>ReleaseStrategy</interfacename> is called it
will be presented with a group containing marked messages (already
processed) and unmarked messages (potentially a new partial
sequence).</para>
is marked as complete. Any new messages for this group will be sent to the discard channel
(if defined). Setting <code>expire-groups-upon-completion</code> to <code>true</code> (default
is <code>false</code>) removes the entire group and any new messages, with the same correlation id
as the removed group, will form a new group.
Partial sequences can be released by using a <classname>MessageGroupStoreReaper</classname>
together with <code>send-partial-result-on-expiry</code> being set to <code>true</code>.</para>
<important>To facilitate discarding of late-arriving messages, the aggregator must maintain state about
the group after it has been released. This can eventually cause out of memory conditions.
To avoid such situations, you should consider configuring a <classname>MessageGroupStoreReaper</classname>
to remove the group metadata; the expiry parameters should be set to expire groups after it is not
expected that late messages will arrive. For information about configuring a reaper, see
<xref linkend="reaper"/>.</important>
<para>Spring Integration provides an out-of-the box implementation for
<interfacename>ReleaseStrategy</interfacename>, the
@@ -338,7 +351,9 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
method="aggregate" ]]><co id="aggxml13" /><![CDATA[
release-strategy="releaseStrategyBean" ]]><co id="aggxml14" /><![CDATA[
release-strategy-method="release"/> ]]><co id="aggxml15" /><![CDATA[
release-strategy-method="release" ]]><co id="aggxml15" /><![CDATA[
expire-groups-upon-completion="false"/> ]]><co id="aggxml16" /><![CDATA[
<int:channel id="outputChannel"/>
@@ -458,6 +473,14 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
present).</emphasis></para>
</callout>
<callout arearefs="aggxml16">
<para>When set to true (default false), completed groups are
removed from the message store, allowing subsequent messages with
the same correlation to form a new group. The default behavior
is to send messages with the same correlation as a completed
group to the <emphasis>discard-channel</emphasis>.</para>
</callout>
</calloutlist>
<para>Using a <code>ref</code> attribute is generally recommended if a custom
@@ -656,7 +679,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
</section>
<section>
<section id="reaper">
<title id="reaper">Managing State in an Aggregator:
MessageGroupStore</title>