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

INT-1129 added receiveAndConvert methods to MessagingOperations and MessagingTemplate

Browse Source
This commit is contained in:
Mark Fisher
2010-08-31 17:49:09 +00:00
parent 4d73334e16
commit 812a8c657f
4 changed files with 219 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -171,8 +171,6 @@ public interface MessagingOperations {
// Convenience methods for receiving auto-converted messages
//-------------------------------------------------------------------------
// TODO: receive and convert methods...
/**
* Receive a message synchronously from the default channel, but only
* wait up to a specified time for delivery. Convert the message into an
@@ -183,7 +181,7 @@ public interface MessagingOperations {
* @return the message received from the channel or <code>null</code> if the timeout expires.
* @throws MessagingException if an error occurs during message reception
*/
//Object receiveAndConvert() throws JmsException;
Object receiveAndConvert() throws MessagingException;
/**
* Receive a message synchronously from the specified channel, but only
@@ -195,7 +193,7 @@ public interface MessagingOperations {
* @return the message received from the channel or <code>null</code> if the timeout expires.
* @throws MessagingException if an error occurs during message reception
*/
//Object receiveAndConvert(PollableChannel channel) throws JmsException;
Object receiveAndConvert(PollableChannel channel) throws MessagingException;
/**
* Receive a message synchronously from the specified channel, but only
@@ -208,7 +206,7 @@ public interface MessagingOperations {
* @return the message received from the channel or <code>null</code> if the timeout expires.
* @throws MessagingException if an error occurs during message reception
*/
//Object receiveAndConvert(String channelName) throws JmsException;
Object receiveAndConvert(String channelName) throws MessagingException;
//-------------------------------------------------------------------------

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

@@ -29,6 +29,7 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.MessagingException;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.channel.ChannelResolutionException;
@@ -269,6 +270,21 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
return this.receive((PollableChannel) channel);
}
public Object receiveAndConvert() throws MessagingException {
Message<?> message = this.receive();
return (message != null) ? this.messageConverter.fromMessage(message) : null;
}
public Object receiveAndConvert(PollableChannel channel) throws MessagingException {
Message<?> message = this.receive(channel);
return (message != null) ? this.messageConverter.fromMessage(message) : null;
}
public Object receiveAndConvert(String channelName) throws MessagingException {
Message<?> message = this.receive(channelName);
return (message != null) ? this.messageConverter.fromMessage(message) : null;
}
public Message<?> sendAndReceive(final Message<?> requestMessage) {
return this.sendAndReceive(this.getRequiredDefaultChannel(), requestMessage);
}

6
spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java
View File

@@ -38,11 +38,13 @@ public class SimpleMessageConverter implements MessageConverter {
}
public SimpleMessageConverter(InboundMessageMapper<?> inboundMessageMapper) {
this(inboundMessageMapper, null);
this(inboundMessageMapper,
(inboundMessageMapper instanceof OutboundMessageMapper ? (OutboundMessageMapper<?>) inboundMessageMapper : null));
}
public SimpleMessageConverter(OutboundMessageMapper<?> outboundMessageMapper) {
this(null, outboundMessageMapper);
this(outboundMessageMapper instanceof InboundMessageMapper ? (InboundMessageMapper<?>) outboundMessageMapper : null,
outboundMessageMapper);
}
public SimpleMessageConverter(InboundMessageMapper<?> inboundMessageMapper, OutboundMessageMapper<?> outboundMessageMapper) {

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

@@ -37,11 +37,15 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.channel.ChannelResolutionException;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.scheduling.support.PeriodicTrigger;
@@ -60,13 +64,7 @@ public class MessagingTemplateTests {
public void setUp() {
this.requestChannel = new QueueChannel();
context.registerChannel("requestChannel", requestChannel);
AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
public Object handleRequestMessage(Message<?> message) {
return message.getPayload().toString().toUpperCase();
}
};
PollingConsumer endpoint = new PollingConsumer(requestChannel, handler);
PollingConsumer endpoint = new PollingConsumer(requestChannel, new TestHandler());
endpoint.setTrigger(new PeriodicTrigger(10));
context.registerEndpoint("testEndpoint", endpoint);
context.refresh();
@@ -218,6 +216,50 @@ public class MessagingTemplateTests {
template.sendAndReceive(new GenericMessage<String>("test"));
}
@Test
public void convertSendAndReceive() {
MessagingTemplate template = new MessagingTemplate();
template.setReceiveTimeout(3000);
Object result = template.convertSendAndReceive(this.requestChannel, "test");
assertNotNull(result);
assertEquals("TEST", result);
}
@Test
public void convertSendAndReceiveWithDefaultChannel() {
MessagingTemplate template = new MessagingTemplate();
template.setDefaultChannel(this.requestChannel);
template.setReceiveTimeout(3000);
Object result = template.convertSendAndReceive("test");
assertNotNull(result);
assertEquals("TEST", result);
}
@Test
public void convertSendAndReceiveWithResolvedChannel() {
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton("testChannel", DirectChannel.class);
context.refresh();
SubscribableChannel testChannel = context.getBean("testChannel", SubscribableChannel.class);
testChannel.subscribe(new TestHandler());
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(context);
template.setReceiveTimeout(3000);
Object result = template.convertSendAndReceive("testChannel", "test");
assertNotNull(result);
assertEquals("TEST", result);
}
@Test(expected = ChannelResolutionException.class)
public void convertSendAndReceiveWithUnresolvableChannel() {
StaticApplicationContext context = new StaticApplicationContext();
context.refresh();
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(context);
template.setReceiveTimeout(3000);
template.convertSendAndReceive("testChannel", "test");
}
@Test
public void sendWithReturnAddress() throws InterruptedException {
final List<String> replies = new ArrayList<String>(3);
@@ -346,4 +388,151 @@ public class MessagingTemplateTests {
template.receive("noSuchChannel");
}
@Test
public void convertAndSendToChannel() {
MessagingTemplate template = new MessagingTemplate();
QueueChannel channel = new QueueChannel();
template.convertAndSend(channel, "test");
Message<?> reply = channel.receive(0);
assertNotNull(reply);
assertEquals("test", reply.getPayload());
}
@Test
public void convertAndSendToDefaultChannel() {
QueueChannel channel = new QueueChannel();
MessagingTemplate template = new MessagingTemplate();
template.setDefaultChannel(channel);
template.convertAndSend("test");
Message<?> reply = channel.receive(0);
assertNotNull(reply);
assertEquals("test", reply.getPayload());
}
@Test
public void convertAndSendToResolvedChannel() {
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton("testChannel", QueueChannel.class);
context.refresh();
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(context);
template.afterPropertiesSet();
template.convertAndSend("testChannel", "test");
PollableChannel channel = context.getBean("testChannel", PollableChannel.class);
Message<?> reply = channel.receive(0);
assertEquals("test", reply.getPayload());
}
@Test(expected = ChannelResolutionException.class)
public void convertAndSendToUnresolvableChannel() {
StaticApplicationContext context = new StaticApplicationContext();
context.refresh();
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(context);
template.afterPropertiesSet();
template.convertAndSend("testChannel", "test");
}
@Test
public void convertAndSendWithCustomConverter() {
MessagingTemplate template = new MessagingTemplate();
TestMapper mapper = new TestMapper();
template.setMessageConverter(new SimpleMessageConverter(mapper, mapper));
QueueChannel channel = new QueueChannel();
template.convertAndSend(channel, "test");
Message<?> reply = channel.receive(0);
assertNotNull(reply);
assertEquals("to:test", reply.getPayload());
}
@Test
public void receiveAndConvertFromChannel() {
MessagingTemplate template = new MessagingTemplate();
QueueChannel channel = new QueueChannel();
channel.send(new GenericMessage<String>("test"));
Object result = template.receiveAndConvert(channel);
assertNotNull(result);
assertEquals("test", result);
}
@Test
public void recieveAndConvertFromDefaultChannel() {
QueueChannel channel = new QueueChannel();
channel.send(new GenericMessage<String>("test"));
MessagingTemplate template = new MessagingTemplate();
template.setDefaultChannel(channel);
Object result = template.receiveAndConvert();
assertNotNull(result);
assertEquals("test", result);
}
@Test
public void receiveAndConvertFromResolvedChannel() {
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton("testChannel", QueueChannel.class);
context.refresh();
PollableChannel channel = context.getBean("testChannel", PollableChannel.class);
channel.send(new GenericMessage<String>("test"));
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(context);
template.afterPropertiesSet();
Object result = template.receiveAndConvert("testChannel");
assertNotNull(result);
assertEquals("test", result);
}
@Test(expected = ChannelResolutionException.class)
public void receiveAndConvertFromUnresolvableChannel() {
StaticApplicationContext context = new StaticApplicationContext();
context.refresh();
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(context);
template.afterPropertiesSet();
template.receiveAndConvert("testChannel");
}
@Test
public void receiveAndConvertWithCustomConverter() {
MessagingTemplate template = new MessagingTemplate();
TestMapper mapper = new TestMapper();
template.setMessageConverter(new SimpleMessageConverter(mapper, mapper));
QueueChannel channel = new QueueChannel();
channel.send(new GenericMessage<String>("test"));
Object result = template.receiveAndConvert(channel);
assertNotNull(result);
assertEquals("from:test", result);
}
@Test
public void convertSendAndReceiveWithCustomConverter() {
TestMapper mapper = new TestMapper();
MessagingTemplate template = new MessagingTemplate();
template.setDefaultChannel(this.requestChannel);
template.setMessageConverter(new SimpleMessageConverter(mapper, mapper));
Object result = template.convertSendAndReceive("test");
assertNotNull(result);
assertEquals("from:TO:TEST", result);
}
private static class TestMapper implements InboundMessageMapper<Object>, OutboundMessageMapper<Object> {
public Object fromMessage(Message<?> message) throws Exception {
return "from:" + message.getPayload();
}
public Message<?> toMessage(Object object) throws Exception {
return new GenericMessage<String>("to:" + object);
}
}
private static class TestHandler extends AbstractReplyProducingMessageHandler {
@Override
public Object handleRequestMessage(Message<?> message) {
return message.getPayload().toString().toUpperCase();
}
}
}