diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index fdda4f8f91..f222c2e641 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -171,7 +171,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName * value indicates that the method should block until either the message is * accepted or the blocking thread is interrupted. */ - public abstract boolean doSend(Message message, long timeout); + protected abstract boolean doSend(Message message, long timeout); /** * Subclasses must implement this method. A non-negative timeout indicates @@ -180,7 +180,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName * indicates that the method should block until either a message is * available or the blocking thread is interrupted. */ - public abstract Message doReceive(long timeout); + protected abstract Message doReceive(long timeout); /** 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 9d82af319f..038dea2227 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 @@ -243,18 +243,17 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA } private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) { - MessageChannel replyChannel = originalMessageHeader.getReplyChannel(); - if (replyChannel != null) { - return replyChannel; + Object returnAddress = originalMessageHeader.getReturnAddress(); + if (returnAddress instanceof MessageChannel) { + return (MessageChannel) returnAddress; } - if (this.channelRegistry == null) { - return null; + if (returnAddress instanceof String && this.channelRegistry != null) { + String channelName = (String) returnAddress; + if (StringUtils.hasText(channelName)) { + return this.channelRegistry.lookupChannel(channelName); + } } - String replyChannelName = originalMessageHeader.getReplyChannelName(); - if (StringUtils.hasText(replyChannelName)) { - return this.channelRegistry.lookupChannel(replyChannelName); - } - if (this.defaultOutputChannelName != null) { + if (this.defaultOutputChannelName != null && this.channelRegistry != null) { return this.channelRegistry.lookupChannel(this.defaultOutputChannelName); } return null; 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 5766827173..47b366128f 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,8 +23,6 @@ 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 @@ -36,23 +34,21 @@ import org.springframework.integration.channel.MessageChannel; */ public class MessageHeader { - private Date timestamp = new Date(); + private final Date timestamp = new Date(); - private Date expiration; + private volatile Date expiration; - private Object correlationId; + private volatile Object correlationId; - private MessageChannel replyChannel; + private volatile Object returnAddress; - private String replyChannelName; + private volatile int sequenceNumber = 1; - private int sequenceNumber = 1; + private volatile int sequenceSize = 1; - private int sequenceSize = 1; + private final Properties properties = new Properties(); - private Properties properties = new Properties(); - - private Map attributes = new ConcurrentHashMap(); + private final Map attributes = new ConcurrentHashMap(); /** @@ -86,20 +82,12 @@ public class MessageHeader { this.correlationId = correlationId; } - public MessageChannel getReplyChannel() { - return this.replyChannel; + public Object getReturnAddress() { + return this.returnAddress; } - public void setReplyChannel(MessageChannel replyChannel) { - this.replyChannel = replyChannel; - } - - public String getReplyChannelName() { - return this.replyChannelName; - } - - public void setReplyChannelName(String replyChannelName) { - this.replyChannelName = replyChannelName; + public void setReturnAddress(Object returnAddress) { + this.returnAddress = returnAddress; } public int getSequenceNumber() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 8891b07bbf..23af0b77be 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -55,7 +55,7 @@ public class MessageBusTests { MessageChannel targetChannel = new SimpleChannel(); bus.registerChannel("sourceChannel", sourceChannel); StringMessage message = new StringMessage("test"); - message.getHeader().setReplyChannelName("targetChannel"); + message.getHeader().setReturnAddress("targetChannel"); sourceChannel.send(message); bus.registerChannel("targetChannel", targetChannel); MessageHandler handler = new MessageHandler() { @@ -105,13 +105,13 @@ public class MessageBusTests { SimpleChannel outputChannel2 = new SimpleChannel(); MessageHandler handler1 = new MessageHandler() { public Message handle(Message message) { - message.getHeader().setReplyChannelName("output1"); + message.getHeader().setReturnAddress("output1"); return message; } }; MessageHandler handler2 = new MessageHandler() { public Message handle(Message message) { - message.getHeader().setReplyChannelName("output2"); + message.getHeader().setReturnAddress("output2"); return message; } }; @@ -137,13 +137,13 @@ public class MessageBusTests { SimpleChannel outputChannel2 = new SimpleChannel(); MessageHandler handler1 = new MessageHandler() { public Message handle(Message message) { - message.getHeader().setReplyChannelName("output1"); + message.getHeader().setReturnAddress("output1"); return message; } }; MessageHandler handler2 = new MessageHandler() { public Message handle(Message message) { - message.getHeader().setReplyChannelName("output2"); + message.getHeader().setReturnAddress("output2"); return message; } }; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 0a0dbcc3b5..8c7eea4412 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -124,7 +124,7 @@ public class EndpointParserTests { ((Lifecycle) endpoint).start(); Message message = new StringMessage("test"); MessageChannel replyChannel = new SimpleChannel(); - message.getHeader().setReplyChannel(replyChannel); + message.getHeader().setReturnAddress(replyChannel); endpoint.handle(message); Message reply = replyChannel.receive(500); assertNotNull(reply); 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 8006f27dc8..d2dbf71246 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 @@ -82,7 +82,7 @@ public class DefaultMessageEndpointTests { endpoint.setHandler(handler); endpoint.start(); StringMessage testMessage = new StringMessage(1, "test"); - testMessage.getHeader().setReplyChannel(replyChannel); + testMessage.getHeader().setReturnAddress(replyChannel); endpoint.handle(testMessage); endpoint.stop(); Message reply = replyChannel.receive(50); @@ -105,7 +105,7 @@ public class DefaultMessageEndpointTests { endpoint.setHandler(handler); endpoint.start(); StringMessage testMessage = new StringMessage(1, "test"); - testMessage.getHeader().setReplyChannelName("replyChannel"); + testMessage.getHeader().setReturnAddress("replyChannel"); endpoint.handle(testMessage); endpoint.stop(); Message reply = replyChannel.receive(50); @@ -114,7 +114,7 @@ public class DefaultMessageEndpointTests { } @Test - public void testReplyChannelTakesPrecedenceOverReplyChannelName() throws Exception { + public void testDynamicReplyChannel() throws Exception { final MessageChannel replyChannel1 = new SimpleChannel(); final MessageChannel replyChannel2 = new SimpleChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); @@ -125,18 +125,25 @@ public class DefaultMessageEndpointTests { } }; DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); endpoint.setHandler(handler); endpoint.start(); - StringMessage testMessage = new StringMessage(1, "test"); - testMessage.getHeader().setReplyChannel(replyChannel1); - testMessage.getHeader().setReplyChannelName("replyChannel2"); + StringMessage testMessage = new StringMessage("test"); + testMessage.getHeader().setReturnAddress(replyChannel1); endpoint.handle(testMessage); - endpoint.stop(); Message reply1 = replyChannel1.receive(50); assertNotNull(reply1); assertEquals("hello test", reply1.getPayload()); Message reply2 = replyChannel2.receive(0); assertNull(reply2); + testMessage.getHeader().setReturnAddress("replyChannel2"); + endpoint.handle(testMessage); + reply1 = replyChannel1.receive(0); + assertNull(reply1); + reply2 = replyChannel2.receive(0); + assertNotNull(reply2); + assertEquals("hello test", reply2.getPayload()); + endpoint.stop(); } @Test @@ -249,7 +256,7 @@ public class DefaultMessageEndpointTests { endpoint.setHandler(new ConcurrentHandler(handler, createExecutor())); endpoint.start(); StringMessage message = new StringMessage(1, "test"); - message.getHeader().setReplyChannelName("replyChannel"); + message.getHeader().setReturnAddress("replyChannel"); endpoint.handle(message); endpoint.stop(); latch.await(500, TimeUnit.MILLISECONDS); @@ -304,7 +311,7 @@ public class DefaultMessageEndpointTests { endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14)); endpoint.start(); StringMessage message = new StringMessage(1, "test"); - message.getHeader().setReplyChannelName("replyChannel"); + message.getHeader().setReturnAddress("replyChannel"); endpoint.handle(message); endpoint.stop(); latch.await(500, TimeUnit.MILLISECONDS); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java index 8493082ad0..35cb578f64 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java @@ -138,7 +138,7 @@ public class AggregatingMessageHandlerTests { message.getHeader().setCorrelationId(correlationId); message.getHeader().setSequenceSize(sequenceSize); message.getHeader().setSequenceNumber(sequenceNumber); - message.getHeader().setReplyChannel(replyChannel); + message.getHeader().setReturnAddress(replyChannel); return message; } @@ -182,7 +182,13 @@ public class AggregatingMessageHandlerTests { try { Message result = this.aggregator.handle(message); if (result != null) { - message.getHeader().getReplyChannel().send(result); + Object returnAddress = message.getHeader().getReturnAddress(); + if (returnAddress instanceof MessageChannel) { + ((MessageChannel) returnAddress).send(result); + } + else { + throw new IllegalStateException("'returnAddress' was not a MessageChannel instance"); + } } } catch (Exception e) {