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

INT-1105: re-organize CorrelatingMessageHandler constructors and formatting

Browse Source
This commit is contained in:
David Syer
2010-05-02 06:11:58 +00:00
parent ee0b597e3b
commit ef31de357a
10 changed files with 68 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,8 +34,8 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* MessageHandler that holds a buffer of correlated messages in a MessageStore. This class takes care of correlated
* groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom
* Message handler that holds a buffer of correlated messages in a {@link MessageStore}. This class takes care of
* correlated groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom
* implementations requiring correlation.
* <p/>
* To customize this handler inject {@link CorrelationStrategy}, {@link ReleaseStrategy}, and
@@ -62,10 +62,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private final MessageGroupProcessor outputProcessor;
private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID);
private volatile CorrelationStrategy correlationStrategy;
private volatile ReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy();
private volatile ReleaseStrategy releaseStrategy;
private MessageChannel outputChannel;
@@ -77,25 +76,24 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private final ConcurrentMap<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
public CorrelatingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy,
ReleaseStrategy ReleaseStrategy, MessageGroupProcessor processor) {
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
Assert.notNull(store);
Assert.notNull(processor);
this.store = store;
this.outputProcessor = processor;
this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID) : correlationStrategy;
this.ReleaseStrategy = ReleaseStrategy == null ? new SequenceSizeReleaseStrategy() : ReleaseStrategy;
this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy;
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
public CorrelatingMessageHandler(MessageStore store, MessageGroupProcessor processor) {
this(store, null, null, processor);
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store) {
this(processor, store, null, null);
}
public CorrelatingMessageHandler(MessageGroupProcessor processor) {
this(new SimpleMessageStore(0), new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID),
new SequenceSizeReleaseStrategy(), processor);
this(processor, new SimpleMessageStore(0), null, null);
}
public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
@@ -103,15 +101,16 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
this.correlationStrategy = correlationStrategy;
}
public void setReleaseStrategy(ReleaseStrategy ReleaseStrategy) {
Assert.notNull(ReleaseStrategy);
this.ReleaseStrategy = ReleaseStrategy;
public void setReleaseStrategy(ReleaseStrategy releaseStrategy) {
Assert.notNull(releaseStrategy);
this.releaseStrategy = releaseStrategy;
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
super.setTaskScheduler(taskScheduler);
}
// TODO: remove unused property setters
public void setTimeout(long timeout) {
}
@@ -157,6 +156,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
message = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build();
}
// TODO: make the lock global?
Object lock = getLock(correlationKey);
synchronized (lock) {
@@ -164,28 +164,36 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
MessageGroup group = new MessageGroup(messages, correlationKey);
if (group.add(message)) {
// TODO: use try/catch to detect problem in group.add() and use
// that to decide on discard?
store(message, correlationKey);
if (ReleaseStrategy.canRelease(group)) {
store(correlationKey, message);
if (releaseStrategy.canRelease(group)) {
if (logger.isDebugEnabled()) {
logger.debug("Completing group with correlationKey [" + correlationKey + "]");
}
outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message,
this.outputChannel));
if (group.isComplete() || group.getSequenceSize()==0) {
complete(group);
if (group.isComplete() || group.getSequenceSize() == 0) {
// The group is complete or else there is no sequence so there is no more state to track
remove(group);
}
else {
partialComplete(group);
// Mark these messages as processed, but do not remove the group from store
mark(group);
}
} // If not releasing any messages the group might still be complete
}
else if (group.isComplete()) {
// If not releasing any messages the group might still be complete
for (Message<?> discard : group.getUnmarked()) {
discardChannel.send(discard);
}
complete(group);
remove(group);
}
}
else {
discardChannel.send(message);
@@ -205,10 +213,10 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
MessageGroup group = new MessageGroup(all, correlationKey);
if (all.size() > 0) {
// last chance for normal completion
if (ReleaseStrategy.canRelease(group)) {
if (releaseStrategy.canRelease(group)) {
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator().next(),
this.outputChannel));
complete(group);
remove(group);
}
else {
if (sendPartialResultOnTimeout) {
@@ -228,7 +236,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
discardChannel.send(message);
}
}
complete(group);
remove(group);
}
return true;
}
@@ -241,19 +249,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
return locks.get(correlationKey);
}
private void partialComplete(MessageGroup group) {
private void mark(MessageGroup group) {
for (Message<?> message : group.getUnmarked()) {
store.mark(group.getCorrelationKey(), message.getHeaders().getId());
}
}
private void complete(MessageGroup group) {
private void remove(MessageGroup group) {
Object correlationKey = group.getCorrelationKey();
store.deleteAll(correlationKey);
locks.remove(correlationKey);
}
private void store(Message<?> message, Object correlationKey) {
private void store(Object correlationKey, Message<?> message) {
store.put(correlationKey, message);
}

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

@@ -53,7 +53,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method.getName());
ReleaseStrategyAdapter ReleaseStrategy = getReleaseStrategy(bean);
CorrelationStrategyAdapter correlationStrategy = getCorrelationStrategy(bean);
CorrelatingMessageHandler handler = new CorrelatingMessageHandler(new SimpleMessageStore(), correlationStrategy, ReleaseStrategy, processor);
CorrelatingMessageHandler handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, ReleaseStrategy);
String discardChannelName = annotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName);

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

@@ -16,14 +16,12 @@
package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the <em>aggregator</em> element of the integration namespace. Registers the annotation-driven
@@ -79,6 +77,8 @@ 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 {
@@ -101,7 +101,6 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
REAPER_INTERVAL_ATTRIBUTE);
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE,

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

@@ -31,27 +31,35 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler");
BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.Resequencer");
IntegrationNamespaceUtils.setValueIfAttributeDefined(processorBuilder, element, "release-partial-sequences");
// TODO: expose message store as an XML attribute
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition());
String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext);
String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder
.getBeanDefinition(), parserContext.getRegistry());
// 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());
String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext);
if (correlationStrategyRef != null) {
builder.addConstructorArgReference(correlationStrategyRef);
}
else {
// Correlation strategy
builder.addConstructorArgReference(processorRef);
}
// Completion strategy
builder.addConstructorArgReference(processorRef);
// Message group processor
builder.addConstructorArgReference(processorRef);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-partial-result-on-timeout");

4
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -48,7 +48,7 @@ public class AggregatorTests {
@Before
public void configureAggregator() {
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor());
this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), new SimpleMessageStore(50));
}
@Test
@@ -211,7 +211,7 @@ public class AggregatorTests {
@Test
public void testNullReturningAggregator() throws InterruptedException {
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor());
this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(50));
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);

8
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -54,8 +54,8 @@ public class ConcurrentAggregatorTests {
@Before
public void configureAggregator() {
this.taskExecutor = new SimpleAsyncTaskExecutor();
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(
50), new MultiplyingProcessor());
this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), new SimpleMessageStore(
50));
}
@Test
@@ -269,8 +269,8 @@ public class ConcurrentAggregatorTests {
@Test
public void testNullReturningAggregator() throws InterruptedException {
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(
50), new NullReturningMessageProcessor());
this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(
50));
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java
View File

@@ -36,7 +36,7 @@ public class CorrelatingMessageHandlerIntegrationTest {
private MessageGroupProcessor processor = new PassThroughMessageGroupProcessor();
private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler(store, processor);
private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler(processor, store);
@Before
public void setupHandler() {

4
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
View File

@@ -65,8 +65,8 @@ public class CorrelatingMessageHandlerTests {
@Before
public void initializeSubject() {
handler = new CorrelatingMessageHandler(new SimpleMessageStore(), correlationStrategy, ReleaseStrategy,
processor);
handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy,
ReleaseStrategy);
handler.setOutputChannel(outputChannel);
doAnswer(new DoesNothing()).when(processor).processAndSend(isA(MessageGroup.class),
isA(MessageChannelTemplate.class), eq(outputChannel));

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -218,6 +218,7 @@ public class MethodInvokingMessageGroupProcessorTests {
testBean = (GreetingService) proxyFactory.getProxy();
MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean);
CorrelatingMessageHandler handler = new CorrelatingMessageHandler(aggregator);
handler.setReleaseStrategy(new MessageCountReleaseStrategy());
handler.setOutputChannel(output);
EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler);
endpoint.start();
@@ -236,6 +237,7 @@ public class MethodInvokingMessageGroupProcessorTests {
testBean = (GreetingService) proxyFactory.getProxy();
MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean);
CorrelatingMessageHandler handler = new CorrelatingMessageHandler(aggregator);
handler.setReleaseStrategy(new MessageCountReleaseStrategy());
handler.setOutputChannel(output);
EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler);
endpoint.start();

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -48,7 +48,7 @@ public class ResequencerTests {
@Before
public void configureResequencer() {
this.resequencer = new CorrelatingMessageHandler(store, processor, processor, processor);
this.resequencer = new CorrelatingMessageHandler(processor, store, processor, processor);
}
@Test