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

Fixes INT-24 (resequencer) by adding a <resequencer/> namespace element.

Browse Source
This commit is contained in:
Marius Bogoevici
2008-06-04 02:34:14 +00:00
parent bfabc1a14b
commit 0f64b81180
7 changed files with 280 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -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<String, Class<? extends BeanDefinitionParser>> parserMappings = this.loadAdapterParserMappings();
try {
for (Map.Entry<String, Class<? extends BeanDefinitionParser>> entry : parserMappings.entrySet()) {

22
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -343,6 +343,28 @@
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="resequencer">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a resequencing message handler.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:attribute name="default-reply-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="discard-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="send-timeout" type="xsd:long" use="optional"/>
<xsd:attribute name="release-partial-sequences" type="xsd:boolean" use="optional"></xsd:attribute>
<xsd:attribute name="send-partial-result-on-timeout" type="xsd:boolean" use="optional"/>
<xsd:attribute name="tracked-correlation-id-capacity" type="xsd:int" use="optional"/>
<xsd:attribute name="reaper-interval" type="xsd:long" use="optional"/>
<xsd:attribute name="timeout" type="xsd:long" use="optional"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="completion-strategy">
<xsd:complexType>

13
org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java
View File

@@ -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());
}
}

65
org.springframework.integration/src/main/java/org/springframework/integration/router/config/ResequencerParser.java Normal file
View File

@@ -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 &gt;resequencer&lt; 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));
}
}
}

137
org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java Normal file
View File

@@ -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<Message<?>> outboundMessages = new ArrayList<Message<?>>();
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 <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel replyChannel) {
GenericMessage<T> message = new GenericMessage<T>(payload);
message.getHeader().setCorrelationId(correlationId);
message.getHeader().setSequenceSize(sequenceSize);
message.getHeader().setSequenceNumber(sequenceNumber);
message.getHeader().setReturnAddress(replyChannel);
return message;
}
}

28
org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<message-bus/>
<channel id="replyChannel" />
<channel id="discardChannel" />
<resequencer id="defaultResequencer"/>
<resequencer id="completelyDefinedResequencer"
default-reply-channel="replyChannel"
discard-channel="discardChannel"
send-timeout="86420000"
send-partial-result-on-timeout="true"
reaper-interval="135"
tracked-correlation-id-capacity="99"
timeout="42"
release-partial-sequences="false"/>
</beans:beans>

31
org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java
View File

@@ -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);