diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index cb4eb8bf1e..e0fce3ea33 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -34,8 +34,8 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; /** - * MessageHandler that holds a buffer of correlated messages in a MessageStore. This class takes care of correlated - * groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom + * Message handler that holds a buffer of correlated messages in a {@link MessageStore}. This class takes care of + * correlated groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom * implementations requiring correlation. *

* To customize this handler inject {@link CorrelationStrategy}, {@link ReleaseStrategy}, and @@ -62,10 +62,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private final MessageGroupProcessor outputProcessor; - private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy( - MessageHeaders.CORRELATION_ID); + private volatile CorrelationStrategy correlationStrategy; - private volatile ReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); + private volatile ReleaseStrategy releaseStrategy; private MessageChannel outputChannel; @@ -77,25 +76,24 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private final ConcurrentMap locks = new ConcurrentHashMap(); - public CorrelatingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy, - ReleaseStrategy ReleaseStrategy, MessageGroupProcessor processor) { + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store, + CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(store); Assert.notNull(processor); this.store = store; this.outputProcessor = processor; this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy( MessageHeaders.CORRELATION_ID) : correlationStrategy; - this.ReleaseStrategy = ReleaseStrategy == null ? new SequenceSizeReleaseStrategy() : ReleaseStrategy; + this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy; this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); } - public CorrelatingMessageHandler(MessageStore store, MessageGroupProcessor processor) { - this(store, null, null, processor); + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store) { + this(processor, store, null, null); } public CorrelatingMessageHandler(MessageGroupProcessor processor) { - this(new SimpleMessageStore(0), new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID), - new SequenceSizeReleaseStrategy(), processor); + this(processor, new SimpleMessageStore(0), null, null); } public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { @@ -103,15 +101,16 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements this.correlationStrategy = correlationStrategy; } - public void setReleaseStrategy(ReleaseStrategy ReleaseStrategy) { - Assert.notNull(ReleaseStrategy); - this.ReleaseStrategy = ReleaseStrategy; + public void setReleaseStrategy(ReleaseStrategy releaseStrategy) { + Assert.notNull(releaseStrategy); + this.releaseStrategy = releaseStrategy; } public void setTaskScheduler(TaskScheduler taskScheduler) { super.setTaskScheduler(taskScheduler); } + // TODO: remove unused property setters public void setTimeout(long timeout) { } @@ -157,6 +156,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements message = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build(); } + // TODO: make the lock global? Object lock = getLock(correlationKey); synchronized (lock) { @@ -164,28 +164,36 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageGroup group = new MessageGroup(messages, correlationKey); if (group.add(message)) { - // TODO: use try/catch to detect problem in group.add() and use - // that to decide on discard? - store(message, correlationKey); - if (ReleaseStrategy.canRelease(group)) { + + store(correlationKey, message); + + if (releaseStrategy.canRelease(group)) { + if (logger.isDebugEnabled()) { logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message, this.outputChannel)); - if (group.isComplete() || group.getSequenceSize()==0) { - complete(group); + if (group.isComplete() || group.getSequenceSize() == 0) { + // The group is complete or else there is no sequence so there is no more state to track + remove(group); } else { - partialComplete(group); + // Mark these messages as processed, but do not remove the group from store + mark(group); } - } // If not releasing any messages the group might still be complete + + } else if (group.isComplete()) { + + // If not releasing any messages the group might still be complete for (Message discard : group.getUnmarked()) { discardChannel.send(discard); } - complete(group); + remove(group); + } + } else { discardChannel.send(message); @@ -205,10 +213,10 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageGroup group = new MessageGroup(all, correlationKey); if (all.size() > 0) { // last chance for normal completion - if (ReleaseStrategy.canRelease(group)) { + if (releaseStrategy.canRelease(group)) { outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator().next(), this.outputChannel)); - complete(group); + remove(group); } else { if (sendPartialResultOnTimeout) { @@ -228,7 +236,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements discardChannel.send(message); } } - complete(group); + remove(group); } return true; } @@ -241,19 +249,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements return locks.get(correlationKey); } - private void partialComplete(MessageGroup group) { + private void mark(MessageGroup group) { for (Message message : group.getUnmarked()) { store.mark(group.getCorrelationKey(), message.getHeaders().getId()); } } - private void complete(MessageGroup group) { + private void remove(MessageGroup group) { Object correlationKey = group.getCorrelationKey(); store.deleteAll(correlationKey); locks.remove(correlationKey); } - private void store(Message message, Object correlationKey) { + private void store(Object correlationKey, Message message) { store.put(correlationKey, message); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index 93b067fd16..dfa2f104c6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -53,7 +53,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method.getName()); ReleaseStrategyAdapter ReleaseStrategy = getReleaseStrategy(bean); CorrelationStrategyAdapter correlationStrategy = getCorrelationStrategy(bean); - CorrelatingMessageHandler handler = new CorrelatingMessageHandler(new SimpleMessageStore(), correlationStrategy, ReleaseStrategy, processor); + CorrelatingMessageHandler handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, ReleaseStrategy); String discardChannelName = annotation.discardChannel(); if (StringUtils.hasText(discardChannelName)) { MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index 2ac76b98ec..38cb2f29c7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -16,14 +16,12 @@ package org.springframework.integration.config.xml; -import org.w3c.dom.Element; - -import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Parser for the aggregator element of the integration namespace. Registers the annotation-driven @@ -79,6 +77,8 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor").getBeanDefinition()); } + // TODO: expose message store as an XML attribute + if (innerHandlerDefinition != null) { processorBuilder.addConstructorArgValue(innerHandlerDefinition); } else { @@ -101,7 +101,6 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE); -// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE); this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE, diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java index 83f8408d8b..6d4f1ed3ab 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java @@ -31,27 +31,35 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler"); BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition( IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.Resequencer"); IntegrationNamespaceUtils.setValueIfAttributeDefined(processorBuilder, element, "release-partial-sequences"); - // TODO: expose message store as an XML attribute - builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition()); - String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext); + String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder .getBeanDefinition(), parserContext.getRegistry()); + + // Message group processor + builder.addConstructorArgReference(processorRef); + + // TODO: expose message store as an XML attribute + // Message store + builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition()); + + String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext); if (correlationStrategyRef != null) { builder.addConstructorArgReference(correlationStrategyRef); } else { + // Correlation strategy builder.addConstructorArgReference(processorRef); } // Completion strategy builder.addConstructorArgReference(processorRef); - // Message group processor - builder.addConstructorArgReference(processorRef); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-partial-result-on-timeout"); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 286ea051ee..ba86bd6da8 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -48,7 +48,7 @@ public class AggregatorTests { @Before public void configureAggregator() { - this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor()); + this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), new SimpleMessageStore(50)); } @Test @@ -211,7 +211,7 @@ public class AggregatorTests { @Test public void testNullReturningAggregator() throws InterruptedException { - this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor()); + this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(50)); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index 8ae1082f86..9eb9b3ff1c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -54,8 +54,8 @@ public class ConcurrentAggregatorTests { @Before public void configureAggregator() { this.taskExecutor = new SimpleAsyncTaskExecutor(); - this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore( - 50), new MultiplyingProcessor()); + this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), new SimpleMessageStore( + 50)); } @Test @@ -269,8 +269,8 @@ public class ConcurrentAggregatorTests { @Test public void testNullReturningAggregator() throws InterruptedException { - this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore( - 50), new NullReturningMessageProcessor()); + this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore( + 50)); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java index 73d83e5362..98e9385828 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java @@ -36,7 +36,7 @@ public class CorrelatingMessageHandlerIntegrationTest { private MessageGroupProcessor processor = new PassThroughMessageGroupProcessor(); - private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler(store, processor); + private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler(processor, store); @Before public void setupHandler() { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 9d35cc87c2..03f2fbdcf5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -65,8 +65,8 @@ public class CorrelatingMessageHandlerTests { @Before public void initializeSubject() { - handler = new CorrelatingMessageHandler(new SimpleMessageStore(), correlationStrategy, ReleaseStrategy, - processor); + handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, + ReleaseStrategy); handler.setOutputChannel(outputChannel); doAnswer(new DoesNothing()).when(processor).processAndSend(isA(MessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index abbeee2188..cbf0b65d44 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -218,6 +218,7 @@ public class MethodInvokingMessageGroupProcessorTests { testBean = (GreetingService) proxyFactory.getProxy(); MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean); CorrelatingMessageHandler handler = new CorrelatingMessageHandler(aggregator); + handler.setReleaseStrategy(new MessageCountReleaseStrategy()); handler.setOutputChannel(output); EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler); endpoint.start(); @@ -236,6 +237,7 @@ public class MethodInvokingMessageGroupProcessorTests { testBean = (GreetingService) proxyFactory.getProxy(); MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean); CorrelatingMessageHandler handler = new CorrelatingMessageHandler(aggregator); + handler.setReleaseStrategy(new MessageCountReleaseStrategy()); handler.setOutputChannel(output); EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler); endpoint.start(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index c8aee832c5..1251675726 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -48,7 +48,7 @@ public class ResequencerTests { @Before public void configureResequencer() { - this.resequencer = new CorrelatingMessageHandler(store, processor, processor, processor); + this.resequencer = new CorrelatingMessageHandler(processor, store, processor, processor); } @Test