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 ecee0a38ae..c4e7c432dc 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 @@ -58,7 +58,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements public static final long DEFAULT_TIMEOUT = 60000L; - private final MessageGroupStore store; + private MessageGroupStore messageStore; private final MessageGroupProcessor outputProcessor; @@ -80,7 +80,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(store); Assert.notNull(processor); - this.store = store; + this.messageStore = store; store.registerExpiryCallback(new MessageGroupCallback() { public void execute(MessageGroup group) { forceComplete(group); @@ -100,6 +100,10 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements public CorrelatingMessageHandler(MessageGroupProcessor processor) { this(processor, new SimpleMessageStore(0), null, null); } + + public void setMessageStore(MessageGroupStore messageStore) { + this.messageStore = messageStore; + } public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { Assert.notNull(correlationStrategy); @@ -149,7 +153,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements Object lock = getLock(correlationKey); synchronized (lock) { - MessageGroup group = store.getMessageGroup(correlationKey); + MessageGroup group = messageStore.getMessageGroup(correlationKey); if (group.add(message)) { @@ -238,17 +242,17 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } private void mark(MessageGroup group) { - store.markMessageGroup(group); + messageStore.markMessageGroup(group); } private void remove(MessageGroup group) { Object correlationKey = group.getCorrelationKey(); - store.removeMessageGroup(correlationKey); + messageStore.removeMessageGroup(correlationKey); locks.remove(correlationKey); } private void store(Object correlationKey, Message message) { - store.addMessageToGroup(correlationKey, message); + messageStore.addMessageToGroup(correlationKey, message); } } 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 ad39b259d6..0bba1d487a 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 @@ -42,6 +42,8 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method"; + private static final String MESSAGE_STORE_ATTRIBUTE = "message-store"; + private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel"; private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; @@ -86,6 +88,8 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + MESSAGE_STORE_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 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 167192e8a4..068baec22b 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 @@ -26,6 +26,7 @@ import org.w3c.dom.Element; * Parser for the <resequencer> element. * * @author Marius Bogoevici + * @author Dave Syer */ public class ResequencerParser extends AbstractConsumerEndpointParser { @@ -60,6 +61,7 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { // Release strategy builder.addConstructorArgReference(processorRef); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-store"); 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/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index c47d7121e4..2874e3fe6a 100644 --- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -1749,6 +1749,21 @@ + + + + Reference to a MessageGroupStore for holding state in between message processing. The default + is to use a volatile in-memory store, which means that unprocessed messages will be lost if the + JVM exits. To customize the expiry of incomplete message groups configure the message store. + + + + + + + + @@ -1792,6 +1807,21 @@ + + + + Reference to a MessageGroupStore for holding state in between message processing. The default + is to use a volatile in-memory store, which means that unprocessed messages will be lost if the + JVM exits. To customize the expiry of incomplete message groups configure the message store. + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests-context.xml new file mode 100644 index 0000000000..b48cac6230 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests-context.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests.java new file mode 100644 index 0000000000..6c70db05b0 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2002-2010 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 static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class AggregatorWithMessageStoreParserTests { + + @Autowired + @Qualifier("input") + private MessageChannel input; + + @Autowired + private TestAggregatorBean aggregatorBean; + + @Autowired + private MessageGroupStore messageGroupStore; + + @Test + public void testAggregation() { + + input.send(createMessage("123", "id1", 3, 1, null)); + assertEquals(1, messageGroupStore.getMessageGroup("id1").size()); + input.send(createMessage("789", "id1", 3, 3, null)); + assertEquals(2, messageGroupStore.getMessageGroup("id1").size()); + input.send(createMessage("456", "id1", 3, 2, null)); + assertEquals("One and only one message should have been aggregated", 1, aggregatorBean + .getAggregatedMessages().size()); + Message aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1"); + assertEquals("The aggregated message payload is not correct", "123456789", aggregatedMessage + .getPayload()); + } + + + private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel outputChannel) { + return MessageBuilder.withPayload(payload) + .setCorrelationId(correlationId) + .setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber) + .setReplyChannel(outputChannel).build(); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests-context.xml new file mode 100644 index 0000000000..0273554938 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests-context.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests.java new file mode 100644 index 0000000000..272cf508e9 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests.java @@ -0,0 +1,84 @@ +/* + * Copyright 2002-2010 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ResequencerWithMessageStoreParserTests { + + @Autowired + @Qualifier("input") + private MessageChannel input; + + @Autowired + @Qualifier("output") + private PollableChannel output; + + @Autowired + private MessageGroupStore messageGroupStore; + + @Test + public void testResequence() { + + input.send(createMessage("123", "id1", 3, 1, null)); + assertEquals(1, messageGroupStore.getMessageGroup("id1").size()); + input.send(createMessage("789", "id1", 3, 3, null)); + assertEquals(2, messageGroupStore.getMessageGroup("id1").size()); + input.send(createMessage("456", "id1", 3, 2, null)); + + Message message1 = output.receive(500); + Message message2 = output.receive(500); + Message message3 = output.receive(500); + + assertNotNull(message1); + assertEquals(new Integer(1), message1.getHeaders().getSequenceNumber()); + assertNotNull(message2); + assertEquals(new Integer(2), message2.getHeaders().getSequenceNumber()); + assertNotNull(message3); + assertEquals(new Integer(3), message3.getHeaders().getSequenceNumber()); + + } + + + private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel outputChannel) { + return MessageBuilder.withPayload(payload) + .setCorrelationId(correlationId) + .setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber) + .setReplyChannel(outputChannel).build(); + } + +}