diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java index 7931fe0a42..b06f397391 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java @@ -22,6 +22,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.util.Assert; @@ -38,13 +39,15 @@ public class MessagePublishingInterceptor implements MethodInterceptor { protected final Log logger = LogFactory.getLog(getClass()); - private volatile MessageChannel defaultChannel; + private volatile MessageChannel outputChannel; + + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); private volatile PayloadType payloadType = PayloadType.RETURN_VALUE; - public void setDefaultChannel(MessageChannel defaultChannel) { - this.defaultChannel = defaultChannel; + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; } public void setPayloadType(PayloadType payloadType) { @@ -82,19 +85,10 @@ public class MessagePublishingInterceptor implements MethodInterceptor { private void sendMessage(Object payload, MethodInvocation invocation) { if (payload != null) { - MessageChannel channel = this.resolveChannel(invocation); - if (channel == null) { - if (logger.isWarnEnabled()) { - logger.warn("unable to resolve channel for intercepted method '" + - invocation.getMethod().getName() + "'"); - } - } - else { - Message message = (payload instanceof Message) - ? (Message) payload - : MessageBuilder.withPayload(payload).build(); - channel.send(message); - } + Message message = (payload instanceof Message) + ? (Message) payload + : MessageBuilder.withPayload(payload).build(); + this.channelTemplate.send(message, this.resolveChannel(invocation)); } } @@ -102,7 +96,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor { * Subclasses may override this method to provide custom behavior. */ protected MessageChannel resolveChannel(MethodInvocation invocation) { - return this.defaultChannel; + return this.outputChannel; } protected PayloadType determinePayloadType(MethodInvocation invocation) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java index c355f12c6b..b021c69871 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java @@ -25,9 +25,6 @@ public interface ChannelRegistry { static final String ERROR_CHANNEL_NAME = "errorChannel"; - - void registerChannel(MessageChannel channel); - MessageChannel lookupChannel(String channelName); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java index 26992dbe2a..fce2715765 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java @@ -35,7 +35,7 @@ public class MessagePublishingInterceptorTests { public void testNonNullReturnValuePublishedWithDefaultChannel() { QueueChannel channel = new QueueChannel(); MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); - interceptor.setDefaultChannel(channel); + interceptor.setOutputChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), interceptor); proxy.messageTest(); Message message = channel.receive(0); @@ -47,7 +47,7 @@ public class MessagePublishingInterceptorTests { public void testNullReturnValueNotPublished() { QueueChannel channel = new QueueChannel(); MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); - interceptor.setDefaultChannel(channel); + interceptor.setOutputChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl(null), interceptor); proxy.messageTest(); assertNull(channel.receive(0)); @@ -57,7 +57,7 @@ public class MessagePublishingInterceptorTests { public void testVoidReturnValueNotPublished() { QueueChannel channel = new QueueChannel(); MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); - interceptor.setDefaultChannel(channel); + interceptor.setOutputChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl(null), interceptor); proxy.voidTest(); assertNull(channel.receive(0)); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java index 46359dbdfe..18d83ff285 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java @@ -41,7 +41,7 @@ public class PublisherAnnotationAdvisorTests { public void testPublisherAnnotation() { final QueueChannel channel = new QueueChannel(); channel.setBeanName("testChannel"); - ChannelRegistry channelRegistry = new TestChannelRegistry(); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); channelRegistry.registerChannel(channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); @@ -55,7 +55,7 @@ public class PublisherAnnotationAdvisorTests { public void testNoPublisherAnnotation() { final QueueChannel channel = new QueueChannel(); channel.setBeanName("testChannel"); - ChannelRegistry channelRegistry = new TestChannelRegistry(); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); channelRegistry.registerChannel(channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); @@ -68,7 +68,7 @@ public class PublisherAnnotationAdvisorTests { public void testPublishArguments() { final QueueChannel channel = new QueueChannel(); channel.setBeanName("testChannel"); - ChannelRegistry channelRegistry = new TestChannelRegistry(); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); channelRegistry.registerChannel(channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); @@ -86,7 +86,7 @@ public class PublisherAnnotationAdvisorTests { public void testPublishException() { final QueueChannel channel = new QueueChannel(); channel.setBeanName("testChannel"); - ChannelRegistry channelRegistry = new TestChannelRegistry(); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); channelRegistry.registerChannel(channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); @@ -109,7 +109,7 @@ public class PublisherAnnotationAdvisorTests { public void testPublishReturnValue() { final QueueChannel channel = new QueueChannel(); channel.setBeanName("testChannel"); - ChannelRegistry channelRegistry = new TestChannelRegistry(); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); channelRegistry.registerChannel(channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java index d3bab333e6..40287f6197 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java @@ -22,13 +22,14 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; @@ -90,7 +91,7 @@ public class ServiceActivatorEndpointTests { public void returnAddressHeaderWithChannelName() { QueueChannel channel = new QueueChannel(1); channel.setBeanName("testChannel"); - ChannelRegistry channelRegistry = new DefaultMessageBus(); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); channelRegistry.registerChannel(channel); ServiceActivatorEndpoint endpoint = this.createEndpoint(); endpoint.setChannelRegistry(channelRegistry); @@ -113,16 +114,9 @@ public class ServiceActivatorEndpointTests { } }; ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler, "handle"); - endpoint.setChannelRegistry(new ChannelRegistry() { - public MessageChannel lookupChannel(String channelName) { - if (channelName.equals("replyChannel2")) { - return replyChannel2; - } - return null; - } - public void registerChannel(MessageChannel channel) { - } - }); + TestChannelRegistry channelRegistry = new TestChannelRegistry(); + channelRegistry.registerChannel(replyChannel2); + endpoint.setChannelRegistry(channelRegistry); Message testMessage1 = MessageBuilder.withPayload("bar") .setReturnAddress(replyChannel1).build(); endpoint.onMessage(testMessage1); @@ -365,4 +359,18 @@ public class ServiceActivatorEndpointTests { } } + + private static class TestChannelRegistry implements ChannelRegistry { + + private final Map channels = new HashMap(); + + public MessageChannel lookupChannel(String channelName) { + return this.channels.get(channelName); + } + + public void registerChannel(MessageChannel channel) { + this.channels.put(channel.getName(), channel); + } + } + }