Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.

Browse Source
This commit is contained in:
Mark Fisher
2010-07-30 11:48:51 +00:00
parent 7d1e14d9fa
commit 8ffe575b1f
3 changed files with 170 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java
View File

@@ -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: <P> void send(String channelName, Message<P> message) throws MessagingException;
<P> void send(String channelName, Message<P> 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 <code>null</code> if the timeout expires
* @throws ChannelResolutionException if the channel name cannot be resolved
* @throws MessagingException if an error occurs during message reception
*/
// TODO: <P> Message<P> receive(String channelName) throws MessagingException;
<P> Message<P> receive(String channelName) throws MessagingException;
// TODO: receiveSelected(selector), receiveSelected(channel, selector), receiveSelected(channelName, selector) ?

62
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -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.
* <p>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 <P> void send(final String channelName, final Message<P> message) {
this.send(this.resolveChannelName(channelName), message);
}
public <P> Message<P> 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 <P> Message<P> 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 {

107
spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java
View File

@@ -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<String, MessageChannel> channelMap = new HashMap<String, MessageChannel>();
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<String, MessageChannel> channelMap = new HashMap<String, MessageChannel>();
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");
}
}