From e9306a334727657e3f4c6327f04c723e33af6851 Mon Sep 17 00:00:00 2001 From: David Syer Date: Tue, 4 May 2010 06:45:51 +0000 Subject: [PATCH] Merge branch 'work' --- .../aggregator/CorrelatingMessageHandler.java | 12 +++------- .../integration/aggregator/Resequencer.java | 24 +++++++++---------- .../integration/annotation/Aggregator.java | 2 +- .../config/xml/AggregatorParser.java | 2 -- .../config/xml/ResequencerParser.java | 5 ++-- .../aggregator/ResequencerTests.java | 2 +- 6 files changed, 19 insertions(+), 28 deletions(-) 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 de9ef6bf1e..447279f30b 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 @@ -26,7 +26,6 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.handler.AbstractMessageHandler; -import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; @@ -111,7 +110,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements super.setTaskScheduler(taskScheduler); } - // TODO: remove unused property setters + // TODO: INT-958 - remove unused property setters public void setTimeout(long timeout) { } @@ -152,12 +151,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message); } - if (!correlationKey.equals(message.getHeaders().getCorrelationId())) { - // TODO: strategise the treatment of overwritten correlation - message = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build(); - } - - // TODO: make the lock global? + // TODO: INT-1117 - make the lock global? Object lock = getLock(correlationKey); synchronized (lock) { @@ -203,7 +197,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } - // TODO: arrange for this to be called if user desires, e.g. periodically + // TODO: INT-958 - arrange for this to be called if user desires, e.g. periodically public final boolean forceComplete(Object correlationKey) { Object lock = getLock(correlationKey); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java index 686dd5f4a7..00c9106f7a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java @@ -28,23 +28,27 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.store.MessageGroup; /** - * This class implements all the strategy interfaces needed for a default - * resequencer. + * This class implements all the strategy interfaces needed for a default resequencer. * * @author Iwein Fuld * @author Dave Syer + * * @since 2.0 */ -public class Resequencer implements CorrelationStrategy, ReleaseStrategy, MessageGroupProcessor { +public class Resequencer implements ReleaseStrategy, MessageGroupProcessor { private volatile SequenceNumberComparator sequenceNumberComparator = new SequenceNumberComparator(); private volatile boolean releasePartialSequences; - public Object getCorrelationKey(Message message) { - // TODO: remove this (as its duplicating the default) - Object correlationKey = message.getHeaders().getCorrelationId(); - return correlationKey; + /** + * Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be + * ordered they will be released, provided they all have sequence numbers greater than those already released. + * + * @param releasePartialSequences + */ + public void setReleasePartialSequences(boolean releasePartialSequences) { + this.releasePartialSequences = releasePartialSequences; } public boolean canRelease(MessageGroup messages) { @@ -69,14 +73,10 @@ public class Resequencer implements CorrelationStrategy, ReleaseStrategy, Messag } } - public void setReleasePartialSequences(boolean releasePartialSequences) { - this.releasePartialSequences = releasePartialSequences; - } - private static class SequenceNumberComparator implements Comparator> { public int compare(Message o1, Message o2) { return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber()); } } - + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java index 0cd416c365..a8d30d2345 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java @@ -78,7 +78,7 @@ public @interface Aggregator { * may be recognized as belonging to an aggregate that has already completed * or timed out */ - // TODO: remove / deal with tracked id capacity + // TODO: INT-958 - remove / deal with tracked id capacity int trackedCorrelationIdCapacity() default 42; } 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 38cb2f29c7..2b267e9b85 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 @@ -77,8 +77,6 @@ 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 { 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 6d4f1ed3ab..ea8d1b1c4e 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 @@ -44,7 +44,6 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { // 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()); @@ -55,9 +54,9 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { } else { // Correlation strategy - builder.addConstructorArgReference(processorRef); + builder.addConstructorArgValue(null); } - // Completion strategy + // Release strategy builder.addConstructorArgReference(processorRef); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel"); 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 845188ec33..b797f4986b 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(processor, store, processor, processor); + this.resequencer = new CorrelatingMessageHandler(processor, store, null, processor); } @Test