diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index 66d912689e..ed8fb5fdfa 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -34,6 +34,7 @@ import org.springframework.integration.channel.config.RendezvousChannelParser; import org.springframework.integration.channel.config.ThreadLocalChannelParser; import org.springframework.integration.config.annotation.AnnotationDrivenParser; import org.springframework.integration.gateway.config.GatewayParser; +import org.springframework.integration.router.config.ResequencerParser; import org.springframework.integration.router.config.RouterParser; import org.springframework.integration.router.config.SplitterParser; import org.springframework.util.ClassUtils; @@ -72,6 +73,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("router", new RouterParser()); registerBeanDefinitionParser("splitter", new SplitterParser()); registerBeanDefinitionParser("aggregator", new AggregatorParser()); + registerBeanDefinitionParser("resequencer", new ResequencerParser()); Map> parserMappings = this.loadAdapterParserMappings(); try { for (Map.Entry> entry : parserMappings.entrySet()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index d7b049c72d..0bacbb3bb7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -343,6 +343,28 @@ + + + + + + Defines a resequencing message handler. + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java index 4e384f8bf4..51376f7099 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java @@ -38,14 +38,18 @@ import org.springframework.integration.message.Message; */ public class ResequencingMessageHandler extends AbstractMessageBarrierHandler{ - private boolean releasePartialSequences; + private volatile boolean releasePartialSequences = true; - public ResequencingMessageHandler(boolean releasePartialSequences) { - this(null, releasePartialSequences); + public ResequencingMessageHandler() { + this(null); } - public ResequencingMessageHandler(ScheduledExecutorService executor, boolean releasePartialSequences) { + public ResequencingMessageHandler(ScheduledExecutorService executor) { super(executor); + } + + + public void setReleasePartialSequences(boolean releasePartialSequences) { this.releasePartialSequences = releasePartialSequences; } @@ -61,4 +65,5 @@ public class ResequencingMessageHandler extends AbstractMessageBarrierHandler{ return (releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceNumber() == releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceSize()); } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/ResequencerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/ResequencerParser.java new file mode 100644 index 0000000000..3536e42312 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/ResequencerParser.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2008 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. + */ + +package org.springframework.integration.router.config; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.integration.router.ResequencingMessageHandler; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + + +/** + * Parser for the >resequencer< tag. + * @author Marius Bogoevici + */ +public class ResequencerParser extends AbstractSimpleBeanDefinitionParser { + + public static final String DEFAULT_REPLY_CHANNEL_ATTRIBUTE = "default-reply-channel"; + + public static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; + + private static final String DEFAULT_REPLY_CHANNEL_PROPERTY = "defaultReplyChannel"; + + private static final String DISCARD_CHANNEL_PROPERTY = "discardChannel"; + + + @Override + protected Class getBeanClass(Element element) { + return ResequencingMessageHandler.class; + } + + @Override + protected boolean isEligibleAttribute(String attributeName) { + // TODO Auto-generated method stub + return !DEFAULT_REPLY_CHANNEL_ATTRIBUTE.equals(attributeName) + && !DISCARD_CHANNEL_ATTRIBUTE.equals(attributeName) && super.isEligibleAttribute(attributeName); + } + + @Override + protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { + if (StringUtils.hasText(element.getAttribute(DEFAULT_REPLY_CHANNEL_ATTRIBUTE))) { + beanDefinition.addPropertyReference(DEFAULT_REPLY_CHANNEL_PROPERTY, element + .getAttribute(DEFAULT_REPLY_CHANNEL_ATTRIBUTE)); + } + if (StringUtils.hasText(element.getAttribute(DISCARD_CHANNEL_ATTRIBUTE))) { + beanDefinition.addPropertyReference(DISCARD_CHANNEL_PROPERTY, element + .getAttribute(DISCARD_CHANNEL_ATTRIBUTE)); + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java new file mode 100644 index 0000000000..a2a638be28 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -0,0 +1,137 @@ +/* + * Copyright 2002-2008 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. + */ + +package org.springframework.integration.config; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.router.AggregatingMessageHandler; +import org.springframework.integration.router.CompletionStrategy; +import org.springframework.integration.router.CompletionStrategyAdapter; +import org.springframework.integration.router.ResequencingMessageHandler; +import org.springframework.integration.util.MethodInvoker; + +/** + * @author Marius Bogoevici + */ +public class ResequencerParserTests { + + private ApplicationContext context; + + + @Before + public void setUp() { + this.context = new ClassPathXmlApplicationContext("resequencerParserTests.xml", this.getClass()); + } + + @Test + public void testResequencing() { + ResequencingMessageHandler resequencingHandler = (ResequencingMessageHandler) context + .getBean("defaultResequencer"); + MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel"); + List> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage("123", "id1", 3, 3, replyChannel)); + outboundMessages.add(createMessage("789", "id1", 3, 1, replyChannel)); + outboundMessages.add(createMessage("456", "id1", 3, 2, replyChannel)); + for (Message message : outboundMessages) { + resequencingHandler.handle(message); + } + Message message1 = replyChannel.receive(500); + Message message2 = replyChannel.receive(500); + Message message3 = replyChannel.receive(500); + Assert.assertNotNull(message1); + Assert.assertEquals(1, message1.getHeader().getSequenceNumber()); + Assert.assertNotNull(message2); + Assert.assertEquals(2, message2.getHeader().getSequenceNumber()); + Assert.assertNotNull(message3); + Assert.assertEquals(3, message3.getHeader().getSequenceNumber()); + } + + @Test + public void testDefaultResequencerProperties() { + ResequencingMessageHandler resequencingHandler = (ResequencingMessageHandler) context + .getBean("defaultResequencer"); + DirectFieldAccessor messageHandlerFieldAccessor = new DirectFieldAccessor(resequencingHandler); + Assert.assertNull(messageHandlerFieldAccessor.getPropertyValue("defaultReplyChannel")); + Assert.assertNull(messageHandlerFieldAccessor.getPropertyValue("discardChannel")); + Assert.assertEquals("The ResequencingMessageHandler is not set with the appropriate timeout value", 1000l, + messageHandlerFieldAccessor.getPropertyValue("sendTimeout")); + Assert.assertEquals( + "The ResequencingMessageHandler is not configured with the appropriate 'send partial results on timeout' flag", + false, messageHandlerFieldAccessor.getPropertyValue("sendPartialResultOnTimeout")); + Assert.assertEquals("The ResequencingMessageHandler is not configured with the appropriate reaper interval", + 1000l, messageHandlerFieldAccessor.getPropertyValue("reaperInterval")); + Assert.assertEquals( + "The ResequencingMessageHandler is not configured with the appropriate tracked correlationId capacity", + 1000, messageHandlerFieldAccessor.getPropertyValue("trackedCorrelationIdCapacity")); + Assert.assertEquals("The ResequencingMessageHandler is not configured with the appropriate timeout", + 60000l, messageHandlerFieldAccessor.getPropertyValue("timeout")); + Assert.assertEquals("The ResequencingMessageHandler is not configured with the appropriate 'release partial sequences' flag", + true, messageHandlerFieldAccessor.getPropertyValue("releasePartialSequences")); + } + + @Test + public void testPropertyAssignment() throws Exception { + ResequencingMessageHandler completeResequencingMessageHandler = (ResequencingMessageHandler) context + .getBean("completelyDefinedResequencer"); + MessageChannel defaultReplyChannel = (MessageChannel) context.getBean("replyChannel"); + MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); + DirectFieldAccessor messageHandlerFieldAccessor = new DirectFieldAccessor(completeResequencingMessageHandler); + Assert.assertEquals("The ResequencingMessageHandler is not injected with the appropriate default reply channel", + defaultReplyChannel, messageHandlerFieldAccessor.getPropertyValue("defaultReplyChannel")); + Assert.assertEquals("The ResequencingMessageHandler is not injected with the appropriate discard channel", + discardChannel, messageHandlerFieldAccessor.getPropertyValue("discardChannel")); + Assert.assertEquals("The ResequencingMessageHandler is not set with the appropriate timeout value", 86420000l, + messageHandlerFieldAccessor.getPropertyValue("sendTimeout")); + Assert.assertEquals( + "The ResequencingMessageHandler is not configured with the appropriate 'send partial results on timeout' flag", + true, messageHandlerFieldAccessor.getPropertyValue("sendPartialResultOnTimeout")); + Assert.assertEquals("The ResequencingMessageHandler is not configured with the appropriate reaper interval", + 135l, messageHandlerFieldAccessor.getPropertyValue("reaperInterval")); + Assert.assertEquals( + "The ResequencingMessageHandler is not configured with the appropriate tracked correlationId capacity", + 99, messageHandlerFieldAccessor.getPropertyValue("trackedCorrelationIdCapacity")); + Assert.assertEquals("The ResequencingMessageHandler is not configured with the appropriate timeout", + 42l, messageHandlerFieldAccessor.getPropertyValue("timeout")); + Assert.assertEquals("The ResequencingMessageHandler is not configured with the appropriate 'release partial sequences' flag", + false, messageHandlerFieldAccessor.getPropertyValue("releasePartialSequences")); + } + + private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel replyChannel) { + GenericMessage message = new GenericMessage(payload); + message.getHeader().setCorrelationId(correlationId); + message.getHeader().setSequenceSize(sequenceSize); + message.getHeader().setSequenceNumber(sequenceNumber); + message.getHeader().setReturnAddress(replyChannel); + return message; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml new file mode 100644 index 0000000000..234fd028ce --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + \ No newline at end of file diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java index cd945968f0..077189aca0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java @@ -36,15 +36,16 @@ public class ResequencerMessageHandlerTests { @Test public void testBasicResequencing() throws InterruptedException { - ResequencingMessageHandler aggregator = new ResequencingMessageHandler(false); + ResequencingMessageHandler resequencer = new ResequencingMessageHandler(); + resequencer.setReleasePartialSequences(false); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); Message message3 = createMessage("789", "ABC", 3, 2, replyChannel); CountDownLatch latch = new CountDownLatch(3); - aggregator.handle(message1); - aggregator.handle(message3); - aggregator.handle(message2); + resequencer.handle(message1); + resequencer.handle(message3); + resequencer.handle(message2); latch.await(1000, TimeUnit.MILLISECONDS); Message reply1 = replyChannel.receive(500); Message reply2 = replyChannel.receive(500); @@ -59,16 +60,17 @@ public class ResequencerMessageHandlerTests { @Test public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { - ResequencingMessageHandler aggregator = new ResequencingMessageHandler(true); + ResequencingMessageHandler resequencer = new ResequencingMessageHandler(); + resequencer.setReleasePartialSequences(true); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); CountDownLatch latch = new CountDownLatch(3); - aggregator.handle(message1); - aggregator.handle(message2); - aggregator.handle(message3); + resequencer.handle(message1); + resequencer.handle(message2); + resequencer.handle(message3); latch.await(1000, TimeUnit.MILLISECONDS); Message reply1 = replyChannel.receive(500); Message reply2 = replyChannel.receive(500); @@ -81,7 +83,7 @@ public class ResequencerMessageHandlerTests { assertNull(reply3); // when sending the last message, the whole sequence must have been sent latch = new CountDownLatch(1); - aggregator.handle(message4); + resequencer.handle(message4); latch.await(1000, TimeUnit.MILLISECONDS); reply3 = replyChannel.receive(500); Message reply4 = replyChannel.receive(500); @@ -94,16 +96,17 @@ public class ResequencerMessageHandlerTests { @Test public void testResequencingWithCompleteSequenceRelease() throws InterruptedException { - ResequencingMessageHandler aggregator = new ResequencingMessageHandler(false); + ResequencingMessageHandler resequencer = new ResequencingMessageHandler(); + resequencer.setReleasePartialSequences(false); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); CountDownLatch latch = new CountDownLatch(3); - aggregator.handle(message1); - aggregator.handle(message2); - aggregator.handle(message3); + resequencer.handle(message1); + resequencer.handle(message2); + resequencer.handle(message3); latch.await(1000, TimeUnit.MILLISECONDS); Message reply1 = replyChannel.receive(500); Message reply2 = replyChannel.receive(500); @@ -114,7 +117,7 @@ public class ResequencerMessageHandlerTests { assertNull(reply3); // when sending the last message, the whole sequence must have been sent latch = new CountDownLatch(1); - aggregator.handle(message4); + resequencer.handle(message4); latch.await(1000, TimeUnit.MILLISECONDS); reply1 = replyChannel.receive(500); reply2 = replyChannel.receive(500);