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 c0162053da..efa1c52cd8 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 @@ -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 null 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 null 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 null 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; //------------------------------------------------------------------------- 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 cbdbe8bb8d..191f4bb98c 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 @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java index 5b90380282..b7e9f43256 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java @@ -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) { 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 c11630af38..de30f5ffa0 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 @@ -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("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 replies = new ArrayList(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("test")); + Object result = template.receiveAndConvert(channel); + assertNotNull(result); + assertEquals("test", result); + } + + @Test + public void recieveAndConvertFromDefaultChannel() { + QueueChannel channel = new QueueChannel(); + channel.send(new GenericMessage("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("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("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, OutboundMessageMapper { + + public Object fromMessage(Message message) throws Exception { + return "from:" + message.getPayload(); + } + + public Message toMessage(Object object) throws Exception { + return new GenericMessage("to:" + object); + } + } + + + private static class TestHandler extends AbstractReplyProducingMessageHandler { + + @Override + public Object handleRequestMessage(Message message) { + return message.getPayload().toString().toUpperCase(); + } + } + }