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

INT-1116: add message-store to XML

Browse Source
This commit is contained in:
David Syer
2010-05-06 11:51:38 +00:00
parent 7f72f94cf6
commit c9eee5a011
8 changed files with 250 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

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

4
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -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,

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

@@ -26,6 +26,7 @@ import org.w3c.dom.Element;
* Parser for the &lt;resequencer&gt; 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");

30
org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1749,6 +1749,21 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-store" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.store.MessageGroupStore" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-partial-result-on-timeout" type="xsd:string" />
</xsd:extension>
</xsd:complexContent>
@@ -1792,6 +1807,21 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-store" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.store.MessageGroupStore" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="comparator" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

23
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests-context.xml Normal file
View File

@@ -0,0 +1,23 @@
<?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.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<queue capacity="5" />
</channel>
<channel id="input" />
<aggregator id="aggregator" ref="aggregatorBean"
input-channel="input" output-channel="output" message-store="messageStore"/>
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
<beans:bean id="aggregatorBean"
class="org.springframework.integration.config.TestAggregatorBean" />
</beans:beans>

77
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithMessageStoreParserTests.java Normal file
View File

@@ -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 <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(outputChannel).build();
}
}

20
org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests-context.xml Normal file
View File

@@ -0,0 +1,20 @@
<?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.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="input"/>
<channel id="output">
<queue capacity="5"/>
</channel>
<resequencer id="resequencer" input-channel="input" output-channel="output" message-store="messageStore"/>
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
</beans:beans>

84
org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerWithMessageStoreParserTests.java Normal file
View File

@@ -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 <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(outputChannel).build();
}
}