From 4d73334e162fcc317d5ec409ab1ec1aed39588ed Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 31 Aug 2010 17:06:00 +0000 Subject: [PATCH] INT-1129 added convertAndSend and convertSendAndReceive methods to MessagingOperations and MessagingTemplate --- .../integration/core/MessagingOperations.java | 86 ++++++++++++++++--- .../integration/core/MessagingTemplate.java | 79 ++++++++++++++--- 2 files changed, 143 insertions(+), 22 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 ff805785ed..c0162053da 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 @@ -77,8 +77,6 @@ public interface MessagingOperations { // Convenience methods for sending auto-converted messages //------------------------------------------------------------------------- - // TODO: convert and send methods... - /** * Send the given object to the default channel, converting the object * to a message with a configured MessageConverter. @@ -86,7 +84,7 @@ public interface MessagingOperations { * @param message the object to convert to a message * @throws MessagingException if an error occurs */ - //void convertAndSend(Object message) throws MessagingException; + void convertAndSend(T message) throws MessagingException; /** * Send the given object to the specified channel, converting the object @@ -95,7 +93,7 @@ public interface MessagingOperations { * @param message the object to convert to a message * @throws MessagingException if an error occurs */ - //void convertAndSend(MessageChannel channel, Object message) throws MessagingException; + void convertAndSend(MessageChannel channel, T message) throws MessagingException; /** * Send the given object to the specified channel, converting the object @@ -105,16 +103,16 @@ public interface MessagingOperations { * @param message the object to convert to a message * @throws MessagingException if an error occurs */ - //void convertAndSend(String destinationName, Object message) throws MessagingException; + void convertAndSend(String channelName, T message) throws MessagingException; /** - * Send the given object to the default destination, converting the object - * to a JMS message with a configured MessageConverter. The MessagePostProcessor + * Send the given object to the default channel, converting the object + * to a message with a configured MessageConverter. The MessagePostProcessor * callback allows for modification of the message after conversion. - *

This will only work with a default destination specified! + *

This will only work with a default channel specified! * @param message the object to convert to a message * @param postProcessor the callback to modify the message - * @throws JmsException checked JMSException converted to unchecked + * @throws MessagingException if an error occurs */ //void convertAndSend(Object message, MessagePostProcessor postProcessor) throws MessagingException; @@ -169,8 +167,6 @@ public interface MessagingOperations {

Message

receive(String channelName) throws MessagingException; - // TODO: receiveSelected(selector), receiveSelected(channel, selector), receiveSelected(channelName, selector) ? - //------------------------------------------------------------------------- // Convenience methods for receiving auto-converted messages //------------------------------------------------------------------------- @@ -214,4 +210,72 @@ public interface MessagingOperations { */ //Object receiveAndConvert(String channelName) throws JmsException; + + //------------------------------------------------------------------------- + // Convenience methods for sending request and receiving reply messages + //------------------------------------------------------------------------- + + /** + * Send a message to the default channel and receive a reply. + *

This will only work with a default channel specified! + * @param requestMessage the message to send + * @return the reply Message if received within the receive timeout. + * @throws MessagingException if an error occurs + */ + Message sendAndReceive(Message requestMessage); + + /** + * Send a message to the specified channel and receive a reply. + * @param channel the channel to which the request Message will be sent + * @param requestMessage the message to send + * @return the reply Message if received within the receive timeout. + * @throws MessagingException if an error occurs + */ + Message sendAndReceive(MessageChannel channel, Message requestMessage); + + /** + * Send a message to the specified channel and receive a reply. + * @param channelName the name of the channel to which the request Message will be sent + * (to be resolved to an actual channel by a ChannelResolver) + * @param requestMessage the message to send + * @return the reply Message if received within the receive timeout. + * @throws ChannelResolutionException if the channel name cannot be resolved + * @throws MessagingException if an error occurs + */ + Message sendAndReceive(String channelName, Message requestMessage); + + /** + * Send the given request object to the default channel, converting the object + * to a message with a configured MessageConverter. If a reply Message is + * received within the receive timeout, it will be converted and returned. + *

This will only work with a default channel specified! + * @param request the object to convert to a request message + * @return the result of converting the reply Message + * @throws MessagingException if an error occurs + */ + Object convertSendAndReceive(Object request); + + /** + * Send the given request object to the specified channel, converting the object + * to a message with a configured MessageConverter. If a reply Message is + * received within the receive timeout, it will be converted and returned. + * @param channel the channel to which the request message will be sent + * @param request the object to convert to a request message + * @return the result of converting the reply Message + * @throws MessagingException if an error occurs + */ + Object convertSendAndReceive(MessageChannel channel, Object request); + + /** + * Send the given request object to the specified channel, converting the object + * to a message with a configured MessageConverter. If a reply Message is + * received within the receive timeout, it will be converted and returned. + * @param channelName the name of the channel to which the request message will be sent + * (to be resolved to an actual channel by a ChannelResolver) + * @param request the object to convert to a request message + * @return the result of converting the reply Message + * @throws MessagingException if an error occurs + */ + Object convertSendAndReceive(String channelName, Object request); + } 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 674e73c8bf..cbdbe8bb8d 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 @@ -33,6 +33,8 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.support.channel.ChannelResolutionException; import org.springframework.integration.support.channel.ChannelResolver; +import org.springframework.integration.support.converter.MessageConverter; +import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -59,6 +61,8 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, private volatile ChannelResolver channelResolver; + private volatile MessageConverter messageConverter = new SimpleMessageConverter(); + private volatile long sendTimeout = -1; private volatile long receiveTimeout = -1; @@ -114,6 +118,16 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, this.channelResolver = channelResolver; } + /** + * Set the {@link MessageConverter} that is to be used to convert + * between Messages and objects for this template. + *

The default is {@link SimpleMessageConverter}. + */ + public void setMessageConverter(MessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' must not be null"); + this.messageConverter = messageConverter; + } + /** * Specify the timeout value to use for send operations. * @@ -208,6 +222,27 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, this.send(this.resolveChannelName(channelName), message); } + public void convertAndSend(T object) { + Message message = this.messageConverter.toMessage(object); + if (message != null) { + this.send(message); + } + } + + public void convertAndSend(MessageChannel channel, T object) { + Message message = this.messageConverter.toMessage(object); + if (message != null) { + this.send(channel, message); + } + } + + public void convertAndSend(String channelName, T object) { + Message message = this.messageConverter.toMessage(object); + if (message != null) { + this.send(channelName, message); + } + } + public

Message

receive() { MessageChannel channel = this.getRequiredDefaultChannel(); Assert.state(channel instanceof PollableChannel, @@ -234,20 +269,42 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return this.receive((PollableChannel) channel); } - public Message sendAndReceive(final Message request) { - return this.sendAndReceive(this.getRequiredDefaultChannel(), request); + public Message sendAndReceive(final Message requestMessage) { + return this.sendAndReceive(this.getRequiredDefaultChannel(), requestMessage); } - public Message sendAndReceive(final MessageChannel channel, final Message request) { + public Message sendAndReceive(final MessageChannel channel, final Message requestMessage) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { return txTemplate.execute(new TransactionCallback>() { public Message doInTransaction(TransactionStatus status) { - return doSendAndReceive(channel, request); + return doSendAndReceive(channel, requestMessage); } }); } - return this.doSendAndReceive(channel, request); + return this.doSendAndReceive(channel, requestMessage); + } + + public Message sendAndReceive(final String channelName, final Message requestMessage) { + return this.sendAndReceive(this.resolveChannelName(channelName), requestMessage); + } + + public Object convertSendAndReceive(final Object request) { + Message requestMessage = this.messageConverter.toMessage(request); + Message replyMessage = this.sendAndReceive(requestMessage); + return this.messageConverter.fromMessage(replyMessage); + } + + public Object convertSendAndReceive(final MessageChannel channel, final Object request) { + Message message = this.messageConverter.toMessage(request); + Message reply = this.sendAndReceive(channel, message); + return this.messageConverter.fromMessage(reply); + } + + public Object convertSendAndReceive(final String channelName, final Object request) { + Message message = this.messageConverter.toMessage(request); + Message reply = this.sendAndReceive(channelName, message); + return this.messageConverter.fromMessage(reply); } private void doSend(MessageChannel channel, Message message) { @@ -275,16 +332,16 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return (Message

) message; } - private Message doSendAndReceive(MessageChannel channel, Message request) { - Object originalReplyChannelHeader = request.getHeaders().getReplyChannel(); - Object originalErrorChannelHeader = request.getHeaders().getErrorChannel(); + private Message doSendAndReceive(MessageChannel channel, Message requestMessage) { + Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel(); + Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel(); TemporaryReplyChannel replyChannel = new TemporaryReplyChannel(this.receiveTimeout); - request = MessageBuilder.fromMessage(request) + requestMessage = MessageBuilder.fromMessage(requestMessage) .setReplyChannel(replyChannel) .setErrorChannel(replyChannel) .build(); - this.doSend(channel, request); - Message reply = this.doReceive(replyChannel); + this.doSend(channel, requestMessage); + Message reply = this.doReceive(replyChannel); if (reply != null) { reply = MessageBuilder.fromMessage(reply) .setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)