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

IN PROGRESS - issue INT-574: Chain element should support aggregators mid chain as well as handlers extending AbstractReplyProducingMessageHandler

Browse Source 
http://jira.springframework.org/browse/INT-574

All MessageHandlers with a setOutputChannel will now be accepted in the chain, the existing functionality has remained identical.
This commit is contained in:
Iwein Fuld
2009-02-10 11:54:33 +00:00
parent 3077b73a78
commit 11b0e85afb
2 changed files with 56 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
View File

@@ -18,11 +18,14 @@ package org.springframework.integration.handler;
import java.util.List;
import org.springframework.beans.BeanWrapperImpl;
import org.springframework.beans.PropertyAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.filter.MessageFilter;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
@@ -30,11 +33,37 @@ import org.springframework.util.Assert;
/**
* A composite {@link MessageHandler} implementation that invokes a chain of
* MessageHandler instances in order.
* <p/>
* Each of the handlers has to implement
* <code>public void setOutputChannel(MessageChannel outputChannel);</code>. An
* exception is made for the last handler in the case that the chain itself does
* not have an output channel. No other assumptions about the type of handler
* are made.
* <p/>
* It is expected that each handler will produce reply messages and send them to
* its output channel, although this is not enforced. It is possible to filter
* messages in the middle of the chain, for example using a
* {@link MessageFilter}.
* <p/>
* This component can be used from the namespace to improve the readability of
* the configuration by removing channels that can be created implicitly.
* <p/>
* <pre>
* &lt;chain&gt;
* &lt;filter ref="someFilter"/&gt;
* &lt;bean class="SomeMessageHandlerImplementation"/&gt;
* &lt;transformer ref="someTransformer"/&gt;
* &lt;aggregator ... /&gt;
* &lt;/chain&gt;
* </pre>
*
* @author Mark Fisher
* @author Iwein Fuld
*/
public class MessageHandlerChain extends IntegrationObjectSupport implements MessageHandler {
private static final String OUTPUT_CHANNEL_PROPERTY = "outputChannel";
private volatile List<MessageHandler> handlers;
private volatile MessageChannel outputChannel;
@@ -43,7 +72,6 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
private final Object initializationMonitor = new Object();
public void setHandlers(List<MessageHandler> handlers) {
this.handlers = handlers;
}
@@ -76,32 +104,32 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
boolean first = (i == 0);
boolean last = (i == handlers.size() - 1);
MessageHandler handler = handlers.get(i);
PropertyAccessor accessor = new BeanWrapperImpl(handler);
if (!first) {
EventDrivenConsumer consumer = new EventDrivenConsumer(channel, handler);
consumer.start();
}
if (!last) {
Assert.isInstanceOf(AbstractReplyProducingMessageHandler.class, handler,
"All handlers except for the last one in the chain must implement "
+ AbstractReplyProducingMessageHandler.class.getName());
channel = new DirectChannel();
channel.setBeanName("_" + this + ".channel#" + i);
((AbstractReplyProducingMessageHandler) handler).setOutputChannel(channel);
Assert.notNull(accessor.getPropertyType(OUTPUT_CHANNEL_PROPERTY),
"All handlers except for the last one in the chain must implement property '"
+ OUTPUT_CHANNEL_PROPERTY + "' of type 'MessageChannel'");
accessor.setPropertyValue(OUTPUT_CHANNEL_PROPERTY, channel);
}
else if (handler instanceof AbstractReplyProducingMessageHandler) {
else if (accessor.getPropertyType(OUTPUT_CHANNEL_PROPERTY) != null) {
MessageChannel replyChannel = (this.outputChannel != null) ? this.outputChannel
: new ReplyForwardingMessageChannel();
((AbstractReplyProducingMessageHandler) handler).setOutputChannel(replyChannel);
accessor.setPropertyValue(OUTPUT_CHANNEL_PROPERTY, replyChannel);
}
else {
Assert.isNull(this.outputChannel,
"An output channel was provided, but the final handler in the chain is not an "
+ "instance of [" + AbstractReplyProducingMessageHandler.class.getName() + "]");
"An output channel was provided, but the final handler in the chain does not implement property '"
+ OUTPUT_CHANNEL_PROPERTY + "' of type 'MessageChannel'");
}
}
}
private class ReplyForwardingMessageChannel implements MessageChannel {
public String getName() {
@@ -126,10 +154,10 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
replyChannel = getChannelResolver().resolveChannelName((String) replyChannelHeader);
}
else {
throw new MessageHandlingException(message, "invalid replyChannel type [" + replyChannelHeader.getClass() + "]");
throw new MessageHandlingException(message, "invalid replyChannel type ["
+ replyChannelHeader.getClass() + "]");
}
return (timeout >= 0) ? replyChannel.send(message, timeout) : replyChannel.send(message);
}
}
}

48
org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java
View File

@@ -49,7 +49,7 @@ public class MessageHandlerChainTests {
public void chainWithOutputChannel() {
handler.handleMessage(message);
expectLastCall().times(3);
expect(outputChannel.send(eq(message), anyLong())).andReturn(true);
expect(outputChannel.send(eq(message))).andReturn(true);
replay(allMocks);
List<MessageHandler> handlers = new ArrayList<MessageHandler>();
handlers.add(producer);
@@ -96,7 +96,7 @@ public class MessageHandlerChainTests {
handler.handleMessage(message);
expectLastCall().times(3);
//equality is lost when recreating the message
expect(outputChannel.send(isA(Message.class), anyLong())).andReturn(true);
expect(outputChannel.send(isA(Message.class))).andReturn(true);
replay(allMocks);
List<MessageHandler> handlers = new ArrayList<MessageHandler>();
handlers.add(producer);
@@ -113,8 +113,7 @@ public class MessageHandlerChainTests {
Message<String> message = MessageBuilder.withPayload("test").setReplyChannelName("testChannel").build();
handler.handleMessage(message);
expectLastCall().times(3);
//equality is lost when recreating the message
expect(outputChannel.send(isA(Message.class), anyLong())).andReturn(true);
expect(outputChannel.send(eq(message))).andReturn(true);
replay(allMocks);
DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
beanFactory.registerSingleton("testChannel", outputChannel);
@@ -129,39 +128,24 @@ public class MessageHandlerChainTests {
chain.handleMessage(message);
}
private class ProducingHandlerStub extends AbstractReplyProducingMessageHandler {
private class ProducingHandlerStub implements MessageHandler {
private MessageChannel output;
private final MessageHandler messageHandler;
public ProducingHandlerStub(MessageHandler handler) {
messageHandler = handler;
}
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
messageHandler.handleMessage(requestMessage);
replyMessageHolder.add(requestMessage);
public void setOutputChannel(MessageChannel channel) {
this.output = channel;
}
public void handleMessage(Message<?> message) {
messageHandler.handleMessage(message);
output.send(message);
}
}
// private class ProducingHandlerStub implements MessageHandler {
// private MessageChannel output;
//
// private final MessageHandler messageHandler;
//
// public ProducingHandlerStub(MessageHandler handler) {
// messageHandler = handler;
// }
//
// void setOutputChannel(MessageChannel channel) {
// this.output = channel;
//
// }
//
// public void handleMessage(Message<?> message) {
// messageHandler.handleMessage(message);
// output.send(message);
// }
// }
}