From 753a67b8a476797f64561606fc544de5893850ce Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 31 Jan 2008 23:14:47 +0000 Subject: [PATCH] Added the 'replyChannel' property to MessageHeader. It takes precedence over the 'replyChannelName' when DefaultMessageEndpoint resolves the channel for reply Messages. --- .../endpoint/DefaultMessageEndpoint.java | 16 ++++--- .../integration/message/MessageHeader.java | 12 +++++ .../endpoint/DefaultMessageEndpointTests.java | 46 +++++++++++++++++++ 3 files changed, 68 insertions(+), 6 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index e876cb8eb8..4df62c820b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -228,10 +228,15 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA return null; } - private MessageChannel resolveReplyChannel(String replyChannelName) { + private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) { + MessageChannel replyChannel = originalMessageHeader.getReplyChannel(); + if (replyChannel != null) { + return replyChannel; + } if (this.channelRegistry == null) { return null; } + String replyChannelName = originalMessageHeader.getReplyChannelName(); if (StringUtils.hasText(replyChannelName)) { return this.channelRegistry.lookupChannel(replyChannelName); } @@ -248,12 +253,11 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA if (replyMessage == null) { return; } - String replyChannelName = originalMessageHeader.getReplyChannelName(); - MessageChannel replyChannel = resolveReplyChannel(replyChannelName); + MessageChannel replyChannel = resolveReplyChannel(originalMessageHeader); if (replyChannel == null) { - throw new MessageHandlingException("Unable to determine reply channel for message. " - + "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' " - + "on the message endpoint."); + throw new MessageHandlingException("Unable to determine reply channel for message. " + + "Provide a 'replyChannel' or 'replyChannelName' in the message header " + + "or a 'defaultOutputChannelName' on the message endpoint."); } replyChannel.send(replyMessage); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java index dabc5dd1d0..5766827173 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java @@ -23,6 +23,8 @@ import java.util.Properties; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import org.springframework.integration.channel.MessageChannel; + /** * A holder for Message metadata. This includes information that may be used by * the messaging system (such as correlationId) as well as information @@ -40,6 +42,8 @@ public class MessageHeader { private Object correlationId; + private MessageChannel replyChannel; + private String replyChannelName; private int sequenceNumber = 1; @@ -82,6 +86,14 @@ public class MessageHeader { this.correlationId = correlationId; } + public MessageChannel getReplyChannel() { + return this.replyChannel; + } + + public void setReplyChannel(MessageChannel replyChannel) { + this.replyChannel = replyChannel; + } + public String getReplyChannelName() { return this.replyChannelName; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index dbe4c420de..c1a51dcba9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -70,6 +70,26 @@ public class DefaultMessageEndpointTests { @Test public void testExplicitReplyChannel() throws Exception { + final MessageChannel replyChannel = new SimpleChannel(); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + return new StringMessage("123", "hello " + message.getPayload()); + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setHandler(handler); + endpoint.start(); + StringMessage testMessage = new StringMessage(1, "test"); + testMessage.getHeader().setReplyChannel(replyChannel); + endpoint.handle(testMessage); + endpoint.stop(); + Message reply = replyChannel.receive(50); + assertNotNull(reply); + assertEquals("hello test", reply.getPayload()); + } + + @Test + public void testExplicitReplyChannelName() throws Exception { final MessageChannel replyChannel = new SimpleChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); @@ -91,6 +111,32 @@ public class DefaultMessageEndpointTests { assertEquals("hello test", reply.getPayload()); } + @Test + public void testReplyChannelTakesPrecedenceOverReplyChannelName() throws Exception { + final MessageChannel replyChannel1 = new SimpleChannel(); + final MessageChannel replyChannel2 = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel2", replyChannel2); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + return new StringMessage("123", "hello " + message.getPayload()); + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setHandler(handler); + endpoint.start(); + StringMessage testMessage = new StringMessage(1, "test"); + testMessage.getHeader().setReplyChannel(replyChannel1); + testMessage.getHeader().setReplyChannelName("replyChannel2"); + endpoint.handle(testMessage); + endpoint.stop(); + Message reply1 = replyChannel1.receive(50); + assertNotNull(reply1); + assertEquals("hello test", reply1.getPayload()); + Message reply2 = replyChannel2.receive(0); + assertNull(reply2); + } + @Test public void testCustomErrorHandler() throws InterruptedException { DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();