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

INT-1129 added convertAndSend and convertSendAndReceive methods to MessagingOperations and MessagingTemplate

Browse Source
This commit is contained in:
Mark Fisher
2010-08-31 17:06:00 +00:00
parent 6c5610693a
commit 4d73334e16
2 changed files with 143 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
<T> 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;
<T> 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;
<T> 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.
* <p>This will only work with a default destination specified!
* <p>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 {
<P> Message<P> 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.
* <p>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.
* <p>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);
}

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

@@ -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.
* <p>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 <T> void convertAndSend(T object) {
Message<?> message = this.messageConverter.toMessage(object);
if (message != null) {
this.send(message);
}
}
public <T> void convertAndSend(MessageChannel channel, T object) {
Message<?> message = this.messageConverter.toMessage(object);
if (message != null) {
this.send(channel, message);
}
}
public <T> void convertAndSend(String channelName, T object) {
Message<?> message = this.messageConverter.toMessage(object);
if (message != null) {
this.send(channelName, message);
}
}
public <P> Message<P> 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<Message<?>>() {
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<P>) message;
}
private Message<?> doSendAndReceive(MessageChannel channel, Message<?> request) {
Object originalReplyChannelHeader = request.getHeaders().getReplyChannel();
Object originalErrorChannelHeader = request.getHeaders().getErrorChannel();
private <S, R> Message<R> doSendAndReceive(MessageChannel channel, Message<S> 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<R> reply = this.doReceive(replyChannel);
if (reply != null) {
reply = MessageBuilder.fromMessage(reply)
.setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)