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