From 8c07856fa11691039e4b6e5b75979fc77561eb47 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 30 Jan 2013 15:11:14 +0200 Subject: [PATCH] INT-2899: Remove Redundant Logic from Aggregator Previously, `AggregatingMessageHandler#setExpireGroupsUponCompletion` had additional logic for removing complete MessageGroups. It could produce some overhead on application start-up with big persistent `MessageStore`. This logic played a role to remove empty groups from a `MessageStore`. Since `AbstractCorrelatingMessageHandler#forceComplete` has an ability to to remove empty groups too, this logic became redundant. So, to clean `MessageStore` from empty complete groups, it's sufficient to use a `MessageGroupStoreReaper`. * Remove logic iterating over the `MessageStore` from `AggregatingMessageHandler#setExpireGroupsUponCompletion` * Polishing `AggregatorSupportedUseCasesTests.java` to use `store.expireMessageGroups(0)` * Introduce `empty-group-min-timeout` xml-attribute, populating `AbstractCorrelatingMessageHandler#minimumTimeoutForEmptyGroups` * Add parser tests for `empty-group-min-timeout` attribute JIRA: https://jira.springsource.org/browse/INT-2899 INT-2899: Documentation about changes of ACMH * rename new XSD-attribute to `empty-group-min-timeout` * add to Reference Manual description about `empty-group-min-timeout` * add 'What's new' for `empty-group-min-timeout` * add 2.2-3.0 Migration Guide note INT-2899 Doc Polishing --- .../AbstractCorrelatingMessageHandler.java | 2 +- .../aggregator/AggregatingMessageHandler.java | 36 ++--- ...stractCorrelatingMessageHandlerParser.java | 11 +- .../config/xml/spring-integration-3.0.xsd | 18 +++ .../AggregatorSupportedUseCasesTests.java | 59 +++---- .../config/AggregatorParserTests.java | 19 ++- .../config/ResequencerParserTests.java | 2 + .../config/aggregatorParserTests.xml | 11 +- .../config/resequencerParserTests.xml | 7 +- src/reference/docbook/aggregator.xml | 153 ++++++++++-------- src/reference/docbook/resequencer.xml | 18 ++- src/reference/docbook/whats-new.xml | 19 +++ 12 files changed, 218 insertions(+), 137 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 5fc8c35a69..4df188f150 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -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. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java index 2b615ff4b0..4421497a40 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java @@ -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 messageGroups = this.messageStore.iterator(); - while (messageGroups.hasNext()) { - MessageGroup messageGroup = messageGroups.next(); - if (messageGroup.isComplete()) { - remove(messageGroup); - } - } - } } @Override protected void afterRelease(MessageGroup messageGroup, Collection> 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); } - } + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index a78573d301..da428f6a29 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -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"); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd index 85a1c0ede0..66842061a0 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd @@ -3262,6 +3262,24 @@ is provided, the return value is expected to match a channel name exactly. + + + + 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. + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java index 6295b39daa..b92270ee8a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java @@ -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; } - + } - + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index a6250d85eb..4224f4c9e4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -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) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index f0f83168ab..c486da9dc8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 334591a19b..4dc132d930 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -16,7 +16,7 @@ + input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/> @@ -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']"/> + correlation-strategy-expression="headers['foo']" + empty-group-min-timeout="60000"/> @@ -79,7 +80,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml index e40dfddfe1..834d31c541 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml @@ -29,12 +29,13 @@ - - Implementing an Aggregator requires providing the logic + 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. @@ -34,13 +34,13 @@ In Spring Integration correlation is done by default based on the MessageHeaders.CORRELATION_ID message header. Messages with the same MessageHeaders.CORRELATION_ID 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 CorrelationStrategy (see below). To determine the point at which a group of messages is ready to be processed, a - ReleaseStrategy is consulted. - The default release strategy for the Aggregator will release a group when all + ReleaseStrategy 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 MessageHeaders.SEQUENCE_SIZE header. This default strategy may be overridden by providing a reference to a @@ -121,14 +121,14 @@ }]]> - The CorrelationStrategy is owned by the + The CorrelationStrategy is owned by the AbstractCorrelatingMessageHandler and it has a default value based on the MessageHeaders.CORRELATION_ID message header: When implementing a specific aggregator strategy for an application, a developer can extend AbstractAggregatingMessageGroupProcessor and implement the - aggregatePayloads method. However, there are better solutions, less + aggregatePayloads 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. - + In general, any POJO can implement the aggregation algorithm if it provides a method that - accepts a single java.util.List as an argument + accepts a single java.util.List as an argument (parameterized lists are supported as well). This method will be invoked for aggregating messages as follows: @@ -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. - +
@@ -234,18 +234,18 @@ aggregation, and false otherwise. - + For example: - + >) {...} }]]> ) {...} }]]> @@ -328,7 +328,7 @@ then you should simply provide an implementation of the ReleaseStrate Configuring an Aggregator with XML Spring Integration supports the configuration of an aggregator via - XML through the <aggregator/> element. Below you can see an example + XML through the <aggregator/> element. Below you can see an example of an aggregator. @@ -355,7 +355,8 @@ then you should simply provide an implementation of the ReleaseStrate release-strategy-method="release" ]]> ]]> ]]> @@ -382,7 +383,7 @@ then you should simply provide an implementation of the ReleaseStrate Lifecycle attribute signaling if aggregator should be started during Application Context startup. Optional (default is 'true'). - + The channel from which where aggregator will receive messages. Required. @@ -406,16 +407,16 @@ then you should simply provide an implementation of the ReleaseStrate complete. Optional, by default a volatile in-memory store. - + 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). Optional. - + - 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 MessageGroup is expired (see MessageGroupStore.expireMessageGroups(long)). One way of expiring MessageGroups is by configuring a MessageGroupStoreReaper. However MessageGroups can alternatively be expired by simply calling @@ -426,20 +427,20 @@ then you should simply provide an implementation of the ReleaseStrate Optional. Default - 'false'. - + The timeout interval for sending the aggregated messages to the output or reply channel. Optional. - + - A reference to a bean that implements the message correlation (grouping) - algorithm. The bean can be an implementation of the CorrelationStrategy + A reference to a bean that implements the message correlation (grouping) + algorithm. The bean can be an implementation of the CorrelationStrategy interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use the MessageHeaders.CORRELATION_ID header) . - + A method defined on the bean referenced by correlation-strategy, that implements the @@ -457,7 +458,7 @@ then you should simply provide an implementation of the ReleaseStrate A reference to a bean defined in the application context. The bean must implement the aggregation logic - as described above. Optional (by default the list of aggregated Messages will become a + as described above. Optional (by default the list of aggregated Messages will become a payload of the output message). @@ -465,7 +466,7 @@ then you should simply provide an implementation of the ReleaseStrate that implements the message aggregation algorithm. Optional, depends on ref attribute being defined. - + A reference to a bean that implements the release strategy. The bean can be an implementation of the ReleaseStrategy interface @@ -498,6 +499,20 @@ then you should simply provide an implementation of the ReleaseStrate group to the discard-channel. + + Only applies if a MessageGroupStoreReaper is configured + for the <aggregator>'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 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. 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. + + Using a ref attribute is generally recommended if a custom @@ -570,16 +585,16 @@ then you should simply provide an implementation of the ReleaseStrate strategy method and the aggregator method can be combined in a single bean (all of them or any two). - + Aggregators and Spring Expression Language (SpEL) - + - Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with - SpEL + Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with + SpEL which is recommended if the logic behind such release strategy 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 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. - ]]> - In the above configuration we are using a Collection Projection 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 Collection Projection 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. - + - The same expression-based approach can be applied when dealing with custom Release and + The same expression-based approach can be applied when dealing with custom Release and Correlation strategies. - Instead of defining a bean for a custom CorrelationStrategy via + Instead of defining a bean for a custom CorrelationStrategy via the correlation-strategy attribute, you can implement your simple correlation logic via a SpEL expression and configure it via the correlation-strategy-expression attribute. @@ -618,13 +633,13 @@ then you should simply provide an implementation of the ReleaseStrate For example: - - In the above example it is assumed that the payload has an attribute person with an id + + In the above example it is assumed that the payload has an attribute person with an id which is going to be used to correlate messages. - Likewise, for the ReleaseStrategy you can implement your release logic as - a SpEL expression and configure it via the release-strategy-expression attribute. + Likewise, for the ReleaseStrategy you can implement your release logic as + a SpEL expression and configure it via the release-strategy-expression 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 #this within the expression. @@ -632,9 +647,9 @@ then you should simply provide an implementation of the ReleaseStrate For example: - + In this example the root object of the SpEL Evaluation Context is the - MessageGroup itself, and you are simply stating + MessageGroup itself, and you are simply stating that as soon as there are more than 5 messages in this group, it should be released.
@@ -645,9 +660,9 @@ then you should simply provide an implementation of the ReleaseStrate An aggregator configured using annotations would look like this. items) { ... } @@ -693,7 +708,7 @@ then you should simply provide an implementation of the ReleaseStrate the @MessageEndpoint is defined on the class, detected automatically through classpath scanning. - +
@@ -709,8 +724,8 @@ then you should simply provide an implementation of the ReleaseStrate All state is carried by the MessageGroup and its management is delegated to the MessageGroupStore. - - + + ReleaseStrate int expireMessageGroups(long timeout); }]]> - + For more information please refer to the JavaDoc. - + The MessageGroupStore accumulates state information in MessageGroups 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 ReleaseStrate - + ]]> The reaper is a Runnable, 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. - + - It is important to understand that the 'timeout' property of the MessageGroupStoreReaper 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 MessageGroupStoreReaper task. For example if - the timeout is set for 10 min, but the MessageGroupStoreReaper task is scheduled to run every 60 min - and the last execution of the MessageGroupStoreReaper task happened 1 min before the timeout, the + It is important to understand that the 'timeout' property of the MessageGroupStoreReaper 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 MessageGroupStoreReaper task. For example if + the timeout is set for 10 min, but the MessageGroupStoreReaper task is scheduled to run every 60 min + and the last execution of the MessageGroupStoreReaper task happened 1 min before the timeout, the MessageGroup 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. In addition to the reaper, the expiry callbacks are invoked when the application - shuts down via a lifecycle callback in the CorrelatingMessageHandler. + shuts down via a lifecycle callback in the AbstractCorrelatingMessageHandler. - The CorrelatingMessageHandler registers its + The AbstractCorrelatingMessageHandler registers its own expiry callback, and this is the link with the boolean flag send-partial-result-on-expiry 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. + + + When using a MessageGroupStoreReaper, it is generally recommended to use a + separate MessageStore for each correlating endpoint. Otherwise, + unexpected results may occur because one endpoint may remove another endpoint's groups. + Some MessageStore implementations allow using the same physical + resources, by partitioning the data; for example, the JdbcMessageStore + has a region property; + the MongoDbMessageStore has a collectionName property. + For more information about MessageStore interface + and its implementations, please read . +
diff --git a/src/reference/docbook/resequencer.xml b/src/reference/docbook/resequencer.xml index 177e18f141..0c02fa46dd 100644 --- a/src/reference/docbook/resequencer.xml +++ b/src/reference/docbook/resequencer.xml @@ -52,7 +52,8 @@ release-strategy="releaseStrategyBean" ]]>]]> + release-strategy-expression="size() == 10" ]]>]]> @@ -170,6 +171,21 @@ release-strategy or release-strategy-expression is allowed. + + + Only applies if a MessageGroupStoreReaper is configured + for the <resequcencer>'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 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. 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. + + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 36167fc614..4fdc765862 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -10,4 +10,23 @@ were resolved as part of the 3.0 development process. +
+ New Components +
+ +
+ General Changes + +
+ Aggregator 'empty-group-min-timeout' property + AbstractCorrelatingMessageHandler provides a new property + empty-group-min-timeout + to allow empty group expiry to run on a longer schedule than expiring partial groups. Empty groups will + not be removed from the MessageStore until they have not been modified + for at least this number of milliseconds. For more information see . + +
+ +
+