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

Merge pull request #727 from artembilan/INT-2899

Browse Source
This commit is contained in:
Gary Russell
2013-02-12 14:29:29 -05:00
parent 22859bf63e 8c07856fa1
commit 6eb009fd62
12 changed files with 218 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -178,7 +178,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
* 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
* late-arriving messages. If you wish to expire empty groups 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.

36
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java
View File

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
@@ -14,17 +14,18 @@
package org.springframework.integration.aggregator;
import java.util.Collection;
import java.util.Iterator;
import org.springframework.integration.Message;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
/**
* Aggregator specific implementation of {@link AbstractCorrelatingMessageHandler}.
* Will remove {@link MessageGroup}s only if 'expireGroupsUponCompletion' flag is set to 'true'.
* Aggregator specific implementation of {@link AbstractCorrelatingMessageHandler}.
* Will remove {@link MessageGroup}s in the {@linkplain #afterRelease}
* only if 'expireGroupsUponCompletion' flag is set to 'true'.
*
* @author Oleg Zhurakousky
* @author Artem Bilan
* @since 2.1
*/
public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler {
@@ -39,33 +40,24 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler
public AggregatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) {
super(processor, store);
}
public AggregatingMessageHandler(MessageGroupProcessor processor) {
super(processor);
}
/**
* Will set the 'expireGroupsUponCompletion' flag and if it is
* set to 'true' it will also remove all 'complete' {@link MessageGroup}s
* @param expireGroupsUponCompletion
* Will set the 'expireGroupsUponCompletion' flag
*
* @see #afterRelease
*/
public void setExpireGroupsUponCompletion(boolean expireGroupsUponCompletion) {
this.expireGroupsUponCompletion = expireGroupsUponCompletion;
if (expireGroupsUponCompletion) {
Iterator<MessageGroup> messageGroups = this.messageStore.iterator();
while (messageGroups.hasNext()) {
MessageGroup messageGroup = messageGroups.next();
if (messageGroup.isComplete()) {
remove(messageGroup);
}
}
}
}
@Override
protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
this.messageStore.completeGroup(messageGroup.getGroupId());
if (this.expireGroupsUponCompletion) {
remove(messageGroup);
}
@@ -73,7 +65,7 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler
for (Message<?> message : messageGroup.getMessages()) {
this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message);
}
}
}
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java
View File

@@ -25,6 +25,7 @@ import org.w3c.dom.Element;
*
* @author Oleg Zhurakousky
* @author Stefan Ferstl
* @author Artem Bilan
* @since 2.1
*
*/
@@ -66,6 +67,7 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "empty-group-min-timeout", "minimumTimeoutForEmptyGroups");
}
protected void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute,
@@ -86,7 +88,7 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
BeanMetadataElement adapter = null;
if (hasBeanRef) {
adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass, parserContext);
adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass);
}
else if (hasExpression) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
@@ -96,16 +98,15 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
adapter = adapterBuilder.getBeanDefinition();
}
else if (processor != null) {
adapter = this.createAdapter(processor, beanMethod, adapterClass, parserContext);
adapter = this.createAdapter(processor, beanMethod, adapterClass);
}
else {
adapter = this.createAdapter(null, beanMethod, adapterClass, parserContext);
adapter = this.createAdapter(null, beanMethod, adapterClass);
}
builder.addPropertyValue(beanProperty, adapter);
}
private BeanMetadataElement createAdapter(BeanMetadataElement ref, String method, String unqualifiedClassName,
ParserContext parserContext) {
private BeanMetadataElement createAdapter(BeanMetadataElement ref, String method, String unqualifiedClassName) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".config." + unqualifiedClassName
+ "FactoryBean");

18
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
View File

@@ -3262,6 +3262,24 @@ is provided, the return value is expected to match a channel name exactly.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="empty-group-min-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Only applies if a MessageGroupStoreReaper is configured for this Correlation
Endpoint's MessageStore.
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.
Note that the actual time to expire an
empty group will also be affected by the reaper's 'timeout'
property and it could be as much as this value plus the timeout.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

59
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java
View File

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
@@ -31,14 +31,15 @@ import static org.junit.Assert.assertNull;
/**
* @author Oleg Zhurakousky
* @author Artem Bilan
*
*/
public class AggregatorSupportedUseCasesTests {
private MessageGroupStore store = new SimpleMessageStore(100);
private DefaultAggregatingMessageGroupProcessor processor = new DefaultAggregatingMessageGroupProcessor();
private AggregatingMessageHandler defaultHandler = new AggregatingMessageHandler(processor, store);
@Test
@@ -47,25 +48,25 @@ public class AggregatorSupportedUseCasesTests {
QueueChannel discardChannel = new QueueChannel();
defaultHandler.setOutputChannel(outputChannel);
defaultHandler.setDiscardChannel(discardChannel);
for (int i = 0; i < 5; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setSequenceSize(5).setCorrelationId("A").setSequenceNumber(i).build());
}
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertNull(discardChannel.receive(0));
assertEquals(0, store.getMessageGroup("A").getMessages().size());
// send another message with the same correlation id and see it in the discard channel
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setSequenceSize(5).setCorrelationId("A").setSequenceNumber(3).build());
assertNotNull(discardChannel.receive(0));
// set 'expireGroupsUponCompletion' to 'true' and the messages should start accumulating again
defaultHandler.setExpireGroupsUponCompletion(true);
// expireMessageGroups from aggregator MessageStore and the messages should start accumulating again
store.expireMessageGroups(0);
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setSequenceSize(5).setCorrelationId("A").setSequenceNumber(3).build());
assertNull(discardChannel.receive(0));
assertEquals(1, store.getMessageGroup("A").getMessages().size());
}
@Test
public void waitForAllCustomReleaseStrategyWithLateArrivals(){
QueueChannel outputChannel = new QueueChannel();
@@ -73,25 +74,25 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setOutputChannel(outputChannel);
defaultHandler.setDiscardChannel(discardChannel);
defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
for (int i = 0; i < 5; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
}
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertNull(discardChannel.receive(0));
assertEquals(0, store.getMessageGroup("A").getMessages().size());
// send another message with the same correlation id and see it in the discard channel
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("A").build());
assertNotNull(discardChannel.receive(0));
// set 'expireGroupsUponCompletion' to 'true' and the messages should start accumulating again
defaultHandler.setExpireGroupsUponCompletion(true);
// expireMessageGroups from aggregator MessageStore and the messages should start accumulating again
store.expireMessageGroups(0);
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("A").build());
assertNull(discardChannel.receive(0));
assertEquals(1, store.getMessageGroup("A").getMessages().size());
}
@Test
public void firstBest(){
QueueChannel outputChannel = new QueueChannel();
@@ -99,7 +100,7 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setOutputChannel(outputChannel);
defaultHandler.setDiscardChannel(discardChannel);
defaultHandler.setReleaseStrategy(new FirstBestReleaseStrategy());
for (int i = 0; i < 5; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
}
@@ -109,7 +110,7 @@ public class AggregatorSupportedUseCasesTests {
assertNotNull(discardChannel.receive(0));
assertNotNull(discardChannel.receive(0));
}
@Test
public void batchingWithoutLeftovers(){
QueueChannel outputChannel = new QueueChannel();
@@ -118,7 +119,7 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setDiscardChannel(discardChannel);
defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
defaultHandler.setExpireGroupsUponCompletion(true);
for (int i = 0; i < 10; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
}
@@ -126,7 +127,7 @@ public class AggregatorSupportedUseCasesTests {
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertNull(discardChannel.receive(0));
}
@Test
public void batchingWithLeftovers(){
QueueChannel outputChannel = new QueueChannel();
@@ -135,7 +136,7 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setDiscardChannel(discardChannel);
defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
defaultHandler.setExpireGroupsUponCompletion(true);
for (int i = 0; i < 12; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
}
@@ -144,21 +145,21 @@ public class AggregatorSupportedUseCasesTests {
assertNull(discardChannel.receive(0));
assertEquals(2, store.getMessageGroup("A").getMessages().size());
}
private class SampleSizeReleaseStrategy implements ReleaseStrategy {
public boolean canRelease(MessageGroup group) {
return group.getMessages().size() == 5;
}
}
private class FirstBestReleaseStrategy implements ReleaseStrategy {
public boolean canRelease(MessageGroup group) {
return true;
}
}
}

19
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,12 @@
package org.springframework.integration.config;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -49,18 +55,13 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
/**
* @author Marius Bogoevici
* @author Mark Fisher
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class AggregatorParserTests {
@@ -192,7 +193,7 @@ public class AggregatorParserTests {
Assert.assertNotNull(reply);
assertEquals(11l, reply.getPayload());
}
@Test // see INT-2011
public void testAggregatorWithPojoReleaseStrategyAsCollection() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoReleaseStrategyInputAsCollection");
@@ -232,9 +233,11 @@ public class AggregatorParserTests {
assertSame(context.getBean("aggregatorBean"), messageGroupProcessorTargetObject);
ReleaseStrategy releaseStrategy = (ReleaseStrategy) TestUtils.getPropertyValue(aggregatingMessageHandler, "releaseStrategy");
CorrelationStrategy correlationStrategy = (CorrelationStrategy) TestUtils.getPropertyValue(aggregatingMessageHandler, "correlationStrategy");
Long minimumTimeoutForEmptyGroups = TestUtils.getPropertyValue(aggregatingMessageHandler, "minimumTimeoutForEmptyGroups", Long.class);
assertTrue(ExpressionEvaluatingReleaseStrategy.class.equals(releaseStrategy.getClass()));
assertTrue(ExpressionEvaluatingCorrelationStrategy.class.equals(correlationStrategy.getClass()));
assertEquals(60000L, minimumTimeoutForEmptyGroups.longValue());
}
@Test(expected=BeanDefinitionParsingException.class)

2
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -44,6 +44,7 @@ import org.springframework.integration.test.util.TestUtils;
* @author Dave Syer
* @author Oleg Zhurakousky
* @author Stefan Ferstl
* @author Artem Bilan
*/
public class ResequencerParserTests {
@@ -88,6 +89,7 @@ public class ResequencerParserTests {
true, getPropertyValue(resequencer, "sendPartialResultOnExpiry"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
true, getPropertyValue(getPropertyValue(resequencer, "releaseStrategy"), "releasePartialSequences"));
assertEquals(60000L, getPropertyValue(resequencer, "minimumTimeoutForEmptyGroups", Long.class).longValue());
}
@Test

11
spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml
View File

@@ -16,7 +16,7 @@
<channel id="aggregatorWithReferenceInput"/>
<aggregator id="aggregatorWithReference" ref="aggregatorBean"
input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/>
input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/>
<channel id="completelyDefinedAggregatorInput"/>
<aggregator id="completelyDefinedAggregator"
@@ -25,7 +25,7 @@
discard-channel="discardChannel"
ref="aggregatorBean"
release-strategy="releaseStrategy"
correlation-strategy="correlationStrategy"
correlation-strategy="correlationStrategy"
send-timeout="86420000"
send-partial-result-on-expiry="true"/>
@@ -36,7 +36,7 @@
output-channel="aggregatorWithExpressionsOutput"
expression="?[payload.startsWith('1')].![payload]"
release-strategy-expression="#root.size()>2"
correlation-strategy-expression="headers['foo']"/>
correlation-strategy-expression="headers['foo']"/>
<channel id="aggregatorWithReferenceAndMethodInput"/>
<aggregator id="aggregatorWithReferenceAndMethod"
@@ -68,7 +68,8 @@
input-channel="aggregatorWithExpressionsAndPojoAggregatorInput"
ref="aggregatorBean"
release-strategy-expression="size() == 2"
correlation-strategy-expression="headers['foo']"/>
correlation-strategy-expression="headers['foo']"
empty-group-min-timeout="60000"/>
<beans:bean id="aggregatorBean"
class="org.springframework.integration.config.TestAggregatorBean" />
@@ -79,7 +80,7 @@
<beans:bean id="releaseStrategy"
class="org.springframework.integration.config.TestReleaseStrategy" />
<beans:bean id="correlationStrategy" class="org.springframework.integration.config.TestCorrelationStrategy"/>
<beans:bean id="correlationStrategy" class="org.springframework.integration.config.TestCorrelationStrategy"/>
<beans:bean id="pojoReleaseStrategy"
class="org.springframework.integration.config.MaxValueReleaseStrategy">

7
spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
View File

@@ -29,12 +29,13 @@
<channel id="inputChannel6"/>
<resequencer id="completelyDefinedResequencer"
<resequencer id="completelyDefinedResequencer"
input-channel="inputChannel2"
output-channel="outputChannel"
output-channel="outputChannel"
discard-channel="discardChannel"
send-timeout="86420000"
send-timeout="86420000"
send-partial-result-on-expiry="true"
empty-group-min-timeout="60000"
release-partial-sequences="true"/>
<resequencer id="resequencerWithCorrelationStrategyRefOnly"

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

@@ -25,7 +25,7 @@
Aggregator will create a single message by processing the whole group, and
will send the aggregated message as output.</para>
<para>Implementing an Aggregator requires providing the logic
<para>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.</para>
@@ -34,13 +34,13 @@
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
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>
<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
<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
<code>MessageHeaders.SEQUENCE_SIZE</code> header.
This default strategy may be overridden by providing a reference to a
@@ -121,14 +121,14 @@
}]]></programlisting>
The <interfacename>CorrelationStrategy</interfacename> is owned by the
The <interfacename>CorrelationStrategy</interfacename> is owned by the
<classname>AbstractCorrelatingMessageHandler</classname>
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,
public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
...
this.correlationStrategy = correlationStrategy == null ?
@@ -158,13 +158,13 @@
<para>When implementing a specific aggregator strategy for an application,
a developer can extend
<classname>AbstractAggregatingMessageGroupProcessor</classname> and implement the
<code>aggregatePayloads</code> method. However, there are better solutions, less
<code>aggregatePayloads</code> method. However, there are better solutions, less
coupled to the API, for implementing the aggregation logic which can be configured easily
either through XML or through annotations.</para>
<para>In general, any POJO can implement the
aggregation algorithm if it provides a method that
accepts a single <interfacename>java.util.List</interfacename> as an argument
accepts a single <interfacename>java.util.List</interfacename> as an argument
(parameterized lists are supported as well). This method will be invoked for aggregating
messages as follows:</para>
@@ -194,7 +194,7 @@
implementing the aggregation logic is through a POJO, and using the
XML or annotation support for configuring it in the application.</para>
</note>
</section>
<section>
@@ -234,18 +234,18 @@
aggregation, and false otherwise.</para>
</listitem>
</itemizedlist>
For example:
<programlisting language="java"><![CDATA[public class MyReleaseStrategy {
@ReleaseStrategy
@ReleaseStrategy
public boolean canMessagesBeReleased(List<Message<?>>) {...}
}]]></programlisting>
<programlisting language="java"><![CDATA[public class MyReleaseStrategy {
@ReleaseStrategy
@ReleaseStrategy
public boolean canMessagesBeReleased(List<String>) {...}
}]]></programlisting>
@@ -328,7 +328,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
<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
XML through the <code>&lt;aggregator/&gt;</code> element. Below you can see an example
of an aggregator.</para>
<programlisting lang="xml"><![CDATA[<channel id="inputChannel"/>
@@ -355,7 +355,8 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
release-strategy-method="release" ]]><co id="aggxml16" /><![CDATA[
release-strategy-expression="size() == 5" ]]><co id="aggxml17" /><![CDATA[
expire-groups-upon-completion="false"/> ]]><co id="aggxml18" /><![CDATA[
expire-groups-upon-completion="false" ]]><co id="aggxml18" /><![CDATA[
empty-group-min-timeout="60000" /> ]]><co id="aggxml19" /><![CDATA[
<int:channel id="outputChannel"/>
@@ -382,7 +383,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
<para>Lifecycle attribute signaling if aggregator should be started during Application Context startup.
<emphasis>Optional (default is 'true')</emphasis>.</para>
</callout>
<callout arearefs="aggxml03">
<para>The channel from which where aggregator will receive messages.
<emphasis>Required</emphasis>.</para>
@@ -406,16 +407,16 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
complete. <emphasis>Optional</emphasis>, by default a volatile
in-memory store.</para>
</callout>
<callout arearefs="aggxml07">
<para>Order of this aggregator when more than one handle is subscribed to the same DirectChannel
(use for load balancing purposes).
(use for load balancing purposes).
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="aggxml08">
<para>
Indicates that expired messages should be aggregated and sent to the 'output-channel' or 'replyChannel'
Indicates that expired messages should be aggregated and sent to the 'output-channel' or 'replyChannel'
once their containing <classname>MessageGroup</classname> is expired (see <code>MessageGroupStore.expireMessageGroups(long)</code>).
One way of expiring <classname>MessageGroup</classname>s is by configuring a <classname>MessageGroupStoreReaper</classname>.
However <classname>MessageGroup</classname>s can alternatively be expired by simply calling
@@ -426,20 +427,20 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
<emphasis>Optional</emphasis>.</para>
<para><emphasis>Default - 'false'</emphasis>.</para>
</callout>
<callout arearefs="aggxml09">
<para>The timeout interval for sending the aggregated messages to the output or
reply channel. <emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="aggxml10">
<para>A reference to a bean that implements the message correlation (grouping)
algorithm. The bean can be an implementation of the <interfacename>CorrelationStrategy</interfacename>
<para>A reference to a bean that implements the message correlation (grouping)
algorithm. The bean can be an implementation of the <interfacename>CorrelationStrategy</interfacename>
interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined
as well. <emphasis>Optional (by default, the aggregator will use
the <code>MessageHeaders.CORRELATION_ID</code> header) </emphasis>.</para>
</callout>
<callout arearefs="aggxml11">
<para>A method defined on the bean referenced by
<code>correlation-strategy</code>, that implements the
@@ -457,7 +458,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
<callout arearefs="aggxml13">
<para>A reference to a bean defined in the application context. The bean must implement the aggregation logic
as described above. <emphasis>Optional (by default the list of aggregated Messages will become a
as described above. <emphasis>Optional (by default the list of aggregated Messages will become a
payload of the output message).</emphasis></para>
</callout>
<callout arearefs="aggxml14">
@@ -465,7 +466,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
that implements the message aggregation
algorithm. <emphasis>Optional, depends on <code>ref</code> attribute being defined.</emphasis></para>
</callout>
<callout arearefs="aggxml15">
<para>A reference to a bean that implements the release strategy.
The bean can be an implementation of the <interfacename>ReleaseStrategy</interfacename> interface
@@ -498,6 +499,20 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
group to the <emphasis>discard-channel</emphasis>.</para>
</callout>
<callout arearefs="aggxml19">
<para>Only applies if a <classname>MessageGroupStoreReaper</classname> is configured
for the <code>&lt;aggregator&gt;</code>'s <classname>MessageStore</classname>.
By default, when a <classname>MessageGroupStoreReaper</classname> 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 expire empty groups on a longer
schedule than expiring partial groups, set this property. Empty groups will
then not be removed from the <classname>MessageStore</classname> until they have not been modified
for at least this number of milliseconds. Note that the actual time to expire an
empty group will also be affected by the reaper's <emphasis>timeout</emphasis>
property and it could be as much as this value plus the timeout.</para>
</callout>
</calloutlist>
<para>Using a <code>ref</code> attribute is generally recommended if a custom
@@ -570,16 +585,16 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
strategy method and the aggregator method can be combined in a single
bean (all of them or any two).</para>
</note>
<para>
<emphasis>Aggregators and Spring Expression Language (SpEL)</emphasis>
</para>
<para>
Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with
<ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html">SpEL</ulink>
Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with
<ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html">SpEL</ulink>
which is recommended if the logic behind such <emphasis>release strategy</emphasis> is relatively simple.
Let's say you have a legacy component that was designed to receive an array of objects. We know that the default release
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).
@@ -595,22 +610,22 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
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.
<programlisting language="xml"><![CDATA[<int:aggregator input-channel="aggChannel"
output-channel="replyChannel"
<programlisting language="xml"><![CDATA[<int:aggregator input-channel="aggChannel"
output-channel="replyChannel"
expression="#this.![payload].toArray()"/>]]></programlisting>
In the above configuration we are using a <ulink
url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html#d0e12113">Collection Projection</ulink> expression
to assemble a new collection from the payloads of all messages in the list and then transforming it to an Array, thus
In the above configuration we are using a <ulink
url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html#d0e12113">Collection Projection</ulink> expression
to assemble a new collection from the payloads of all messages in the list and then transforming it to an Array, thus
achieving the same result as the java code above.
</para>
<para>
The same expression-based approach can be applied when dealing with custom <emphasis>Release</emphasis> and
The same expression-based approach can be applied when dealing with custom <emphasis>Release</emphasis> and
<emphasis>Correlation</emphasis> strategies.
</para>
<para>
Instead of defining a bean for a custom <classname>CorrelationStrategy</classname> via
Instead of defining a bean for a custom <classname>CorrelationStrategy</classname> via
the <code>correlation-strategy</code> attribute, you can implement your simple correlation logic
via a SpEL expression and configure it via the <code>correlation-strategy-expression</code> attribute.
</para>
@@ -618,13 +633,13 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
For example:
<programlisting language="xml"><![CDATA[correlation-strategy-expression="payload.person.id"]]></programlisting>
In the above example it is assumed that the payload has an attribute <code>person</code> with an <code>id</code>
In the above example it is assumed that the payload has an attribute <code>person</code> with an <code>id</code>
which is going to be used to correlate messages.
</para>
<para>
Likewise, for the <interfacename>ReleaseStrategy</interfacename> you can implement your release logic as
a SpEL expression and configure it via the <code>release-strategy-expression</code> attribute.
Likewise, for the <interfacename>ReleaseStrategy</interfacename> you can implement your release logic as
a SpEL expression and configure it via the <code>release-strategy-expression</code> attribute.
The only difference is that since ReleaseStrategy is passed the List of Messages, the root object in the SpEL
evaluation context is the List itself. That List can be referenced as <code>#this</code> within the expression.
</para>
@@ -632,9 +647,9 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
For example:
<programlisting language="xml"><![CDATA[release-strategy-expression="#this.size() gt 5"]]></programlisting>
In this example the root object of the SpEL Evaluation Context is the
<interfacename>MessageGroup</interfacename> itself, and you are simply stating
<interfacename>MessageGroup</interfacename> itself, and you are simply stating
that as soon as there are more than 5 messages in this group, it should be released.
</para>
</section>
@@ -645,9 +660,9 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
<para>An aggregator configured using annotations would look like this.</para>
<programlisting language="java"><![CDATA[public class Waiter {
...
...
@Aggregator ]]><co id="aggann" /><![CDATA[
@Aggregator ]]><co id="aggann" /><![CDATA[
public Delivery aggregatingMethod(List<OrderItem> items) {
...
}
@@ -693,7 +708,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
the @MessageEndpoint is defined on the class, detected automatically
through classpath scanning.</para>
</section>
</section>
<section id="reaper">
@@ -709,8 +724,8 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
All state is carried by the <interfacename>MessageGroup</interfacename> and its
management is delegated to the
<interfacename>MessageGroupStore</interfacename>.
<programlisting><![CDATA[public interface MessageGroupStore {
int getMessageCountForAllMessageGroups();
@@ -734,11 +749,11 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
int expireMessageGroups(long timeout);
}]]></programlisting>
For more information please refer to the
<ulink url="http://static.springsource.org/spring-integration/api/org/springframework/integration/store/MessageGroupStore.html">JavaDoc</ulink>.
</para>
<para>The <interfacename>MessageGroupStore</interfacename> accumulates state
information in <interfacename>MessageGroups</interfacename> while waiting for
a release strategy to be triggered, and that event might not ever happen.
@@ -777,32 +792,44 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
<property name="timeout" value="30000"/>
</bean>
<task:scheduled-tasks scheduler="scheduler">
<task:scheduled-tasks scheduler="scheduler">
<task:scheduled ref="reaper" method="run" fixed-rate="10000"/>
</task:scheduled-tasks>]]></programlisting>
<para>The reaper is a <interfacename>Runnable</interfacename>, and all that is happening
in the example above is that the message group store's expire method is being called
once every 10 seconds. The timeout itself is 30 seconds.</para>
<note>
It is important to understand that the 'timeout' property of the <classname>MessageGroupStoreReaper</classname> is an
approximate value and is impacted by the the rate of the task scheduler since this property will
only be checked on the next scheduled execution of the <classname>MessageGroupStoreReaper</classname> task. For example if
the timeout is set for 10 min, but the <classname>MessageGroupStoreReaper</classname> task is scheduled to run every 60 min
and the last execution of the <classname>MessageGroupStoreReaper</classname> task happened 1 min before the timeout, the
It is important to understand that the 'timeout' property of the <classname>MessageGroupStoreReaper</classname> is an
approximate value and is impacted by the the rate of the task scheduler since this property will
only be checked on the next scheduled execution of the <classname>MessageGroupStoreReaper</classname> task. For example if
the timeout is set for 10 min, but the <classname>MessageGroupStoreReaper</classname> task is scheduled to run every 60 min
and the last execution of the <classname>MessageGroupStoreReaper</classname> task happened 1 min before the timeout, the
<classname>MessageGroup</classname> will not expire for the next 59 min. So it is recommended to set the rate at least equal to the value of the timeout or shorter.
</note>
<para>In addition to the reaper, the expiry callbacks are invoked when the application
shuts down via a lifecycle callback in the <classname>CorrelatingMessageHandler</classname>.
shuts down via a lifecycle callback in the <classname>AbstractCorrelatingMessageHandler</classname>.
</para>
<para>The <classname>CorrelatingMessageHandler</classname> registers its
<para>The <classname>AbstractCorrelatingMessageHandler</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, any unmarked messages in groups that are not yet released can
be sent on to the output channel.</para>
<important>
<para>When using a <classname>MessageGroupStoreReaper</classname>, it is generally recommended to use a
separate <classname>MessageStore</classname> for each correlating endpoint. Otherwise,
unexpected results may occur because one endpoint may remove another endpoint's groups.</para>
<para>Some <interfacename>MessageStore</interfacename> implementations allow using the same physical
resources, by partitioning the data; for example, the <classname>JdbcMessageStore</classname>
has a <code>region</code> property;
the <classname>MongoDbMessageStore</classname> has a <code>collectionName</code> property.</para>
<para>For more information about <interfacename>MessageStore</interfacename> interface
and its implementations, please read <xref linkend="message-store"/>.</para>
</important>
</section>
</section>

18
src/reference/docbook/resequencer.xml
View File

@@ -52,7 +52,8 @@
release-strategy="releaseStrategyBean" ]]><co id="resxml14-co" linkends="resxml14" /><![CDATA[
release-strategy-method="release" ]]><co id="resxml15-co" linkends="resxml15" /><![CDATA[
release-strategy-expression="size() == 10" />]]><co id="resxml16-co" linkends="resxml16" /></programlisting>
release-strategy-expression="size() == 10" ]]><co id="resxml16-co" linkends="resxml16" /><![CDATA[
empty-group-min-timeout="60000" />]]><co id="resxml17-co" linkends="resxml17" /></programlisting>
<para><calloutlist>
<callout arearefs="resxml1-co" id="resxml1">
@@ -170,6 +171,21 @@
<code>release-strategy</code>
or <code>release-strategy-expression</code> is allowed.</para>
</callout>
<callout arearefs="resxml17-co" id="resxml17">
<para>Only applies if a <classname>MessageGroupStoreReaper</classname> is configured
for the <code>&lt;resequcencer&gt;</code>'s <classname>MessageStore</classname>.
By default, when a <classname>MessageGroupStoreReaper</classname> 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 expire empty groups on a longer
schedule than expiring partial groups, set this property. Empty groups will
then not be removed from the <classname>MessageStore</classname> until they have not been modified
for at least this number of milliseconds. Note that the actual time to expire an
empty group will also be affected by the reaper's <emphasis>timeout</emphasis>
property and it could be as much as this value plus the timeout.</para>
</callout>
</calloutlist></para>
<note>

19
src/reference/docbook/whats-new.xml
View File

@@ -10,4 +10,23 @@
were resolved as part of the 3.0 development process.
</para>
<section id="3.0-new-components">
<title>New Components</title>
</section>
<section id="3.0-general">
<title>General Changes</title>
<section id="3.0-corr-endpoint-empty-groups">
<title>Aggregator 'empty-group-min-timeout' property</title>
<para><classname>AbstractCorrelatingMessageHandler</classname> provides a new property
<code>empty-group-min-timeout</code>
to allow empty group expiry to run on a longer schedule than expiring partial groups. Empty groups will
not be removed from the <interfacename>MessageStore</interfacename> until they have not been modified
for at least this number of milliseconds. For more information see <xref linkend="aggregator-config"/>.
</para>
</section>
</section>
</chapter>