From 8ffe575b1f7971a60e70ca016519aa9a4b81691d Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 30 Jul 2010 11:48:51 +0000 Subject: [PATCH] INT-1300 MessagingTemplate now provides Channel-resolving send and receive methods (where the 'channelName' can be passed instead of a concrete MessageChannel instance). This temporarily introduces a tangle, but that will be removed in the subsequent refactoring to follow shortly. --- .../integration/core/MessagingOperations.java | 6 +- .../integration/core/MessagingTemplate.java | 62 +++++++++- .../channel/MessagingTemplateTests.java | 107 ++++++++++++++++++ 3 files changed, 170 insertions(+), 5 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java index dabcdfef88..5f3fd0cf79 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java @@ -64,9 +64,10 @@ public interface MessagingOperations { * @param channelName the name of the channel to which the message will be sent * (to be resolved to an actual channel by a ChannelResolver) * @param message the message to send + * @throws ChannelResolutionException if the channel name cannot be resolved * @throws MessagingException if an error occurs during message sending */ - //TODO:

void send(String channelName, Message

message) throws MessagingException; +

void send(String channelName, Message

message) throws MessagingException; //------------------------------------------------------------------------- @@ -159,9 +160,10 @@ public interface MessagingOperations { * @param channelName the name of the channel from which a message should be received * (to be resolved to an actual channel by a ChannelResolver) * @return the message received from the channel or null if the timeout expires + * @throws ChannelResolutionException if the channel name cannot be resolved * @throws MessagingException if an error occurs during message reception */ - // TODO:

Message

receive(String channelName) throws MessagingException; +

Message

receive(String channelName) throws MessagingException; // TODO: receiveSelected(selector), receiveSelected(channel, selector), receiveSelected(channelName, selector) ? diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 4321cc7b16..bf7e4d3ae5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -23,10 +23,13 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHeaders; +import org.springframework.integration.context.BeanFactoryChannelResolver; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -45,12 +48,14 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessagingTemplate implements MessagingOperations, InitializingBean { +public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, InitializingBean { protected final Log logger = LogFactory.getLog(this.getClass()); private volatile MessageChannel defaultChannel; + private volatile ChannelResolver channelResolver; + private volatile long sendTimeout = -1; private volatile long receiveTimeout = -1; @@ -67,6 +72,8 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean private volatile boolean readOnly = false; + private volatile BeanFactory beanFactory; + private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -95,6 +102,17 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean this.defaultChannel = defaultChannel; } + /** + * Set the {@link ChannelResolver} that is to be used to resolve + * {@link MessageChannel} references for this template. + *

When running within an application context, the default resolver is a + * {@link BeanFactoryChannelResolver}. + */ + public void setChannelResolver(ChannelResolver channelResolver) { + Assert.notNull(channelResolver, "'channelResolver' must not be null"); + this.channelResolver = channelResolver; + } + /** * Specify the timeout value to use for send operations. * @@ -145,11 +163,18 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean return this.transactionTemplate; } + public void setBeanFactory(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + public void afterPropertiesSet() { synchronized (this.initializationMonitor) { if (this.initialized) { return; } + if (this.channelResolver == null && this.beanFactory != null) { + this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory); + } if (this.transactionManager != null) { TransactionTemplate template = new TransactionTemplate(this.transactionManager); template.setPropagationBehaviorName(this.propagationBehaviorName); @@ -179,6 +204,10 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean this.doSend(channel, message); } + public

void send(final String channelName, final Message

message) { + this.send(this.resolveChannelName(channelName), message); + } + public

Message

receive() { MessageChannel channel = this.getRequiredDefaultChannel(); Assert.state(channel instanceof PollableChannel, @@ -198,6 +227,13 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean return this.doReceive(channel); } + public

Message

receive(String channelName) { + MessageChannel channel = this.resolveChannelName(channelName); + Assert.isInstanceOf(PollableChannel.class, channel, + "A PollableChannel is required for receive operations. "); + return this.receive((PollableChannel) channel); + } + public Message sendAndReceive(final Message request) { return this.sendAndReceive(this.getRequiredDefaultChannel(), request); } @@ -260,11 +296,31 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean private MessageChannel getRequiredDefaultChannel() { Assert.state(this.defaultChannel != null, - "No 'defaultChannel' specified for MessageChannelTemplate. " - + "Unable to invoke methods without a channel argument."); + "No 'defaultChannel' specified for MessagingTemplate. " + + "Unable to invoke methods without an explicit channel argument."); return this.defaultChannel; } + private ChannelResolver getRequiredChannelResolver() { + Assert.state(this.channelResolver != null, + "No 'channelResolver' specified for MessagingTemplate. " + + "Unable to invoke methods with a channel name argument."); + return this.channelResolver; + } + + /** + * Resolve the given channel name into a {@link MessageChannel}, + * via this template's {@link ChannelResolver} if available. + * @param channelName the name of the channel + * @return the resolved {@link MessageChannel} + * @throws IllegalStateException if this template does not have a ChannelResolver + * @throws ChannelResolutionException if the channel name cannot be resolved + * @see #setChannelResolver + */ + protected MessageChannel resolveChannelName(String channelName) { + return getRequiredChannelResolver().resolveChannelName(channelName); + } + @SuppressWarnings({"unchecked", "unused"}) private static class TemporaryReplyChannel implements PollableChannel { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java index 37aff3dbf0..c9729a0d73 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java @@ -22,7 +22,9 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -30,10 +32,13 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.Message; +import org.springframework.integration.core.ChannelResolutionException; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.core.PollableChannel; import org.springframework.integration.core.StringMessage; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -239,4 +244,106 @@ public class MessagingTemplateTests { assertTrue(replies.contains("TEST3")); } + @Test + public void sendByChannelName() { + StaticApplicationContext context = new StaticApplicationContext(); + context.registerSingleton("testChannel", QueueChannel.class); + context.refresh(); + MessagingTemplate template = new MessagingTemplate(); + template.setBeanFactory(context); + template.afterPropertiesSet(); + Message message = MessageBuilder.withPayload("test").build(); + template.send("testChannel", message); + PollableChannel channel = context.getBean("testChannel", PollableChannel.class); + assertEquals(message, channel.receive(0)); + } + + @Test + public void sendByChannelNameWithCustomChannelResolver() { + QueueChannel testChannel = new QueueChannel(); + Map channelMap = new HashMap(); + channelMap.put("testChannel", testChannel); + MapBasedChannelResolver channelResolver = new MapBasedChannelResolver(channelMap); + MessagingTemplate template = new MessagingTemplate(); + template.setChannelResolver(channelResolver); + template.afterPropertiesSet(); + Message message = MessageBuilder.withPayload("test").build(); + template.send("testChannel", message); + assertEquals(message, testChannel.receive(0)); + } + + @Test(expected = IllegalStateException.class) + public void sendByChannelNameWithoutChannelResolver() { + MessagingTemplate template = new MessagingTemplate(); + template.send("testChannel", MessageBuilder.withPayload("test").build()); + } + + @Test(expected = ChannelResolutionException.class) + public void sendByChannelNameWithUnresolvableChannel() { + StaticApplicationContext context = new StaticApplicationContext(); + context.registerSingleton("testChannel", QueueChannel.class); + context.refresh(); + MessagingTemplate template = new MessagingTemplate(); + template.setBeanFactory(context); + template.afterPropertiesSet(); + Message message = MessageBuilder.withPayload("test").build(); + template.send("noSuchChannel", message); + } + + @Test + public void receiveByChannelName() { + StaticApplicationContext context = new StaticApplicationContext(); + context.registerSingleton("testChannel", QueueChannel.class); + context.refresh(); + MessagingTemplate template = new MessagingTemplate(); + template.setBeanFactory(context); + template.afterPropertiesSet(); + PollableChannel channel = context.getBean("testChannel", PollableChannel.class); + Message message = MessageBuilder.withPayload("test").build(); + channel.send(message); + assertEquals(message, template.receive("testChannel")); + } + + @Test + public void receiveByChannelNameWithCustomChannelResolver() { + QueueChannel testChannel = new QueueChannel(); + Map channelMap = new HashMap(); + channelMap.put("testChannel", testChannel); + MapBasedChannelResolver channelResolver = new MapBasedChannelResolver(channelMap); + MessagingTemplate template = new MessagingTemplate(); + template.setChannelResolver(channelResolver); + template.afterPropertiesSet(); + Message message = MessageBuilder.withPayload("test").build(); + testChannel.send(message); + assertEquals(message, template.receive("testChannel")); + } + + @Test(expected = IllegalArgumentException.class) + public void receiveByChannelNameWithNonPollableChannel() { + StaticApplicationContext context = new StaticApplicationContext(); + context.registerSingleton("testChannel", DirectChannel.class); + context.refresh(); + MessagingTemplate template = new MessagingTemplate(); + template.setBeanFactory(context); + template.afterPropertiesSet(); + template.receive("testChannel"); + } + + @Test(expected = IllegalStateException.class) + public void receiveByChannelNameWithoutChannelResolver() { + MessagingTemplate template = new MessagingTemplate(); + template.receive("testChannel"); + } + + @Test(expected = ChannelResolutionException.class) + public void receiveByChannelNameWithUnresolvableChannel() { + StaticApplicationContext context = new StaticApplicationContext(); + context.registerSingleton("testChannel", QueueChannel.class); + context.refresh(); + MessagingTemplate template = new MessagingTemplate(); + template.setBeanFactory(context); + template.afterPropertiesSet(); + template.receive("noSuchChannel"); + } + }