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

Jms request/reply operations

Browse Source 
This commit updates JmsMessagingTemplate to support the
MessageRequestReplyOperation interface that provides synchronous
request/reply operations.

As JmsMessagingTemplate delegates everything under the scenes to
JmsTemplate, the latter has been updated as well to offer such lower
level operation.

Issue: SPR-12037
This commit is contained in:
Stephane Nicoll
2014-07-28 15:22:31 +02:00
parent c89325b9ca
commit b6389a6c66
9 changed files with 557 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
spring-jms/src/main/java/org/springframework/jms/core/JmsMessageOperations.java
View File

@@ -24,6 +24,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.core.MessageReceivingOperations;
import org.springframework.messaging.core.MessageRequestReplyOperations;
import org.springframework.messaging.core.MessageSendingOperations;
/**
@@ -35,8 +36,8 @@ import org.springframework.messaging.core.MessageSendingOperations;
* @since 4.1
* @see org.springframework.jms.core.JmsTemplate
*/
public interface JmsMessageOperations
extends MessageSendingOperations<Destination>, MessageReceivingOperations<Destination> {
public interface JmsMessageOperations extends MessageSendingOperations<Destination>,
MessageReceivingOperations<Destination>, MessageRequestReplyOperations<Destination> {
/**
* Send a message to the given destination.
@@ -109,4 +110,73 @@ public interface JmsMessageOperations
*/
<T> T receiveAndConvert(String destinationName, Class<T> targetClass) throws MessagingException;
/**
* Send a request message and receive the reply from the given destination.
* @param destinationName the name of the target destination
* @param requestMessage the message to send
* @return the reply, possibly {@code null} if the message could not be received,
* for example due to a timeout
*/
Message<?> sendAndReceive(String destinationName, Message<?> requestMessage) throws MessagingException;
/**
* Convert the given request Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter}, send
* it as a {@link Message} to the given destination, receive the reply and convert
* its body of the specified target class.
* @param destinationName the name of the target destination
* @param request payload for the request message to send
* @param targetClass the target type to convert the payload of the reply to
* @return the payload of the reply message, possibly {@code null} if the message
* could not be received, for example due to a timeout
*/
<T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass) throws MessagingException;
/**
* Convert the given request Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter}, send
* it as a {@link Message} with the given headers, to the specified destination,
* receive the reply and convert its body of the specified target class.
* @param destinationName the name of the target destination
* @param request payload for the request message to send
* @param headers headers for the request message to send
* @param targetClass the target type to convert the payload of the reply to
* @return the payload of the reply message, possibly {@code null} if the message
* could not be received, for example due to a timeout
*/
<T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers, Class<T> targetClass)
throws MessagingException;
/**
* Convert the given request Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* apply the given post processor and send the resulting {@link Message} to the
* given destination, receive the reply and convert its body of the given
* target class.
* @param destinationName the name of the target destination
* @param request payload for the request message to send
* @param targetClass the target type to convert the payload of the reply to
* @param requestPostProcessor post process to apply to the request message
* @return the payload of the reply message, possibly {@code null} if the message
* could not be received, for example due to a timeout
*/
<T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass,
MessagePostProcessor requestPostProcessor) throws MessagingException;
/**
* Convert the given request Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message with the given headers, apply the given post processor
* and send the resulting {@link Message} to the specified destination, receive
* the reply and convert its body of the given target class.
* @param destinationName the name of the target destination
* @param request payload for the request message to send
* @param targetClass the target type to convert the payload of the reply to
* @param requestPostProcessor post process to apply to the request message
* @return the payload of the reply message, possibly {@code null} if the message
* could not be received, for example due to a timeout
*/
<T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
Class<T> targetClass, MessagePostProcessor requestPostProcessor) throws MessagingException;
}

88
spring-jms/src/main/java/org/springframework/jms/core/JmsMessagingTemplate.java
View File

@@ -28,7 +28,7 @@ import org.springframework.jms.support.converter.MessagingMessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.core.AbstractMessageReceivingTemplate;
import org.springframework.messaging.core.AbstractMessagingTemplate;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.util.Assert;
@@ -38,7 +38,7 @@ import org.springframework.util.Assert;
* @author Stephane Nicoll
* @since 4.1
*/
public class JmsMessagingTemplate extends AbstractMessageReceivingTemplate<Destination>
public class JmsMessagingTemplate extends AbstractMessagingTemplate<Destination>
implements JmsMessageOperations, InitializingBean {
private JmsTemplate jmsTemplate;
@@ -206,13 +206,76 @@ public class JmsMessagingTemplate extends AbstractMessageReceivingTemplate<Desti
}
}
@Override
public Message<?> sendAndReceive(Message<?> requestMessage) {
Destination defaultDestination = getDefaultDestination();
if (defaultDestination != null) {
return sendAndReceive(defaultDestination, requestMessage);
}
else {
return sendAndReceive(getRequiredDefaultDestinationName(), requestMessage);
}
}
@Override
public Message<?> sendAndReceive(String destinationName, Message<?> requestMessage)
throws MessagingException {
return doSendAndReceive(destinationName, requestMessage);
}
@Override
public <T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass)
throws MessagingException {
return convertSendAndReceive(destinationName, request, null, targetClass);
}
@Override
public <T> T convertSendAndReceive(Object request, Class<T> targetClass) {
return convertSendAndReceive(request, targetClass, null);
}
@Override
public <T> T convertSendAndReceive(String destinationName, Object request,
Map<String, Object> headers, Class<T> targetClass) throws MessagingException {
return convertSendAndReceive(destinationName, request, headers, targetClass, null);
}
@Override
public <T> T convertSendAndReceive(Object request, Class<T> targetClass, MessagePostProcessor postProcessor) {
Destination defaultDestination = getDefaultDestination();
if (defaultDestination != null) {
return convertSendAndReceive(defaultDestination, request, targetClass, postProcessor);
}
else {
return convertSendAndReceive(getRequiredDefaultDestinationName(), request, targetClass, postProcessor);
}
}
@Override
public <T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass,
MessagePostProcessor requestPostProcessor) throws MessagingException {
return convertSendAndReceive(destinationName, request, null, targetClass, requestPostProcessor);
}
@SuppressWarnings("unchecked")
@Override
public <T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
Class<T> targetClass, MessagePostProcessor postProcessor) {
Message<?> requestMessage = doConvert(request, headers, postProcessor);
Message<?> replyMessage = sendAndReceive(destinationName, requestMessage);
return (replyMessage != null ? (T) getMessageConverter().fromMessage(replyMessage, targetClass) : null);
}
@Override
protected void doSend(Destination destination, Message<?> message) {
this.jmsTemplate.send(destination, new MessagingMessageCreator(message, this.jmsMessageConverter));
this.jmsTemplate.send(destination, createMessageCreator(message));
}
protected void doSend(String destinationName, Message<?> message) {
this.jmsTemplate.send(destinationName, new MessagingMessageCreator(message, this.jmsMessageConverter));
this.jmsTemplate.send(destinationName, createMessageCreator(message));
}
@Override
@@ -226,6 +289,23 @@ public class JmsMessagingTemplate extends AbstractMessageReceivingTemplate<Desti
return doConvert(jmsMessage);
}
@Override
protected Message<?> doSendAndReceive(Destination destination, Message<?> requestMessage) {
javax.jms.Message jmsMessage = this.jmsTemplate
.sendAndReceive(destination, createMessageCreator(requestMessage));
return doConvert(jmsMessage);
}
protected Message<?> doSendAndReceive(String destinationName, Message<?> requestMessage) {
javax.jms.Message jmsMessage = this.jmsTemplate
.sendAndReceive(destinationName, createMessageCreator(requestMessage));
return doConvert(jmsMessage);
}
private MessagingMessageCreator createMessageCreator(Message<?> message) {
return new MessagingMessageCreator(message, this.jmsMessageConverter);
}
protected String getRequiredDefaultDestinationName() {
String name = getDefaultDestinationName();
if (name == null) {

69
spring-jms/src/main/java/org/springframework/jms/core/JmsOperations.java
View File

@@ -32,8 +32,12 @@ import org.springframework.jms.JmsException;
* {@code receive(..)} methods that mirror various JMS API methods.
* See the JMS specification and javadocs for details on those methods.
*
* <p>Provides also basic request reply operation using a temporary
* queue to collect the reply.
*
* @author Mark Pollack
* @author Juergen Hoeller
* @author Stephane Nicoll
* @since 1.1
* @see JmsTemplate
* @see javax.jms.Destination
@@ -83,9 +87,9 @@ public interface JmsOperations {
<T> T execute(String destinationName, ProducerCallback<T> action) throws JmsException;
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for sending messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
/**
* Send a message to the default destination.
@@ -115,9 +119,9 @@ public interface JmsOperations {
void send(String destinationName, MessageCreator messageCreator) throws JmsException;
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for sending auto-converted messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
/**
* Send the given object to the default destination, converting the object
@@ -185,9 +189,9 @@ public interface JmsOperations {
throws JmsException;
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for receiving messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
/**
* Receive a message synchronously from the default destination, but only
@@ -264,9 +268,9 @@ public interface JmsOperations {
Message receiveSelected(String destinationName, String messageSelector) throws JmsException;
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for receiving auto-converted messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
/**
* Receive a message synchronously from the default destination, but only
@@ -349,9 +353,54 @@ public interface JmsOperations {
Object receiveSelectedAndConvert(String destinationName, String messageSelector) throws JmsException;
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for sending messages to and receiving the reply from a destination
//---------------------------------------------------------------------------------------
/**
* Send a request message and receive the reply from a default destination. The
* {@link MessageCreator} callback creates the message given a Session. A temporary
* queue is created as part of this operation and is set in the {@code JMSReplyTO}
* header of the message.
* <p>This will only work with a default destination specified!
* @param messageCreator callback to create a request message
* @return the reply, possibly {@code null} if the message could not be received,
* for example due to a timeout
* @throws JmsException checked JMSException converted to unchecked
*/
Message sendAndReceive(MessageCreator messageCreator) throws JmsException;
/**
* Send a message and receive the reply from the specified destination. The
* {@link MessageCreator} callback creates the message given a Session. A temporary
* queue is created as part of this operation and is set in the {@code JMSReplyTO}
* header of the message.
* @param destination the destination to send this message to
* @param messageCreator callback to create a message
* @return the reply, possibly {@code null} if the message could not be received,
* for example due to a timeout
* @throws JmsException checked JMSException converted to unchecked
*/
Message sendAndReceive(Destination destination, MessageCreator messageCreator) throws JmsException;
/**
* Send a message and receive the reply from the specified destination. The
* {@link MessageCreator} callback creates the message given a Session. A temporary
* queue is created as part of this operation and is set in the {@code JMSReplyTO}
* header of the message.
* @param destinationName the name of the destination to send this message to
* (to be resolved to an actual destination by a DestinationResolver)
* @param messageCreator callback to create a message
* @return the reply, possibly {@code null} if the message could not be received,
* for example due to a timeout
* @throws JmsException checked JMSException converted to unchecked
*/
Message sendAndReceive(String destinationName, MessageCreator messageCreator) throws JmsException;
//---------------------------------------------------------------------------------------
// Convenience methods for browsing messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
/**
* Browse messages in the default JMS queue. The callback gives access to the JMS

125
spring-jms/src/main/java/org/springframework/jms/core/JmsTemplate.java
View File

@@ -28,6 +28,7 @@ import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import org.springframework.jms.JmsException;
import org.springframework.jms.connection.ConnectionFactoryUtils;
@@ -77,6 +78,7 @@ import org.springframework.util.ReflectionUtils;
*
* @author Mark Pollack
* @author Juergen Hoeller
* @author Stephane Nicoll
* @since 1.1
* @see #setConnectionFactory
* @see #setPubSubDomain
@@ -447,9 +449,9 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// JmsOperations execute methods
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
@Override
public <T> T execute(SessionCallback<T> action) throws JmsException {
@@ -546,9 +548,9 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for sending messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
@Override
public void send(MessageCreator messageCreator) throws JmsException {
@@ -635,9 +637,9 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for sending auto-converted messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
@Override
public void convertAndSend(Object message) throws JmsException {
@@ -710,9 +712,9 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for receiving messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
@Override
public Message receive() throws JmsException {
@@ -838,9 +840,9 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for receiving auto-converted messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
@Override
public Object receiveAndConvert() throws JmsException {
@@ -890,9 +892,108 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
// Convenience methods for sending messages to and receiving the reply from a destination
//---------------------------------------------------------------------------------------
@Override
public Message sendAndReceive(MessageCreator messageCreator) throws JmsException {
Destination defaultDestination = getDefaultDestination();
if (defaultDestination != null) {
return sendAndReceive(defaultDestination, messageCreator);
}
else {
return sendAndReceive(getRequiredDefaultDestinationName(), messageCreator);
}
}
@Override
public Message sendAndReceive(final Destination destination, final MessageCreator messageCreator) throws JmsException {
return executeLocal(new SessionCallback<Message>() {
@Override
public Message doInJms(Session session) throws JMSException {
return doSendAndReceive(session, destination, messageCreator);
}
}, true);
}
@Override
public Message sendAndReceive(final String destinationName, final MessageCreator messageCreator) throws JmsException {
return executeLocal(new SessionCallback<Message>() {
@Override
public Message doInJms(Session session) throws JMSException {
Destination destination = resolveDestinationName(session, destinationName);
return doSendAndReceive(session, destination, messageCreator);
}
}, true);
}
/**
* Send a request message to the given {@link Destination} and block until
* a reply has been received on a temporary queue created on-the-fly.
* <p>Return the response message or {@code null} if no message has
* @throws JMSException if thrown by JMS API methods
*/
protected Message doSendAndReceive(Session session, Destination destination, MessageCreator messageCreator)
throws JMSException {
Assert.notNull(messageCreator, "MessageCreator must not be null");
TemporaryQueue responseQueue = null;
MessageProducer producer = null;
MessageConsumer consumer = null;
try {
Message requestMessage = messageCreator.createMessage(session);
responseQueue = session.createTemporaryQueue();
producer = session.createProducer(destination);
consumer = session.createConsumer(responseQueue);
requestMessage.setJMSReplyTo(responseQueue);
if (logger.isDebugEnabled()) {
logger.debug("Sending created message: " + requestMessage);
}
doSend(producer, requestMessage);
return doReceive(consumer, getReceiveTimeout());
}
finally {
JmsUtils.closeMessageConsumer(consumer);
JmsUtils.closeMessageProducer(producer);
if (responseQueue != null) {
responseQueue.delete();
}
}
}
/**
* A variant of {@link #execute(SessionCallback, boolean)} that explicitly
* creates a non transactional session. The given {@link SessionCallback}
* does not participate in an existing transaction.
*/
private <T> T executeLocal(SessionCallback<T> action, boolean startConnection) throws JmsException {
Assert.notNull(action, "Callback object must not be null");
Connection con = null;
Session session = null;
try {
con = getConnectionFactory().createConnection();
session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
if (startConnection) {
con.start();
}
if (logger.isDebugEnabled()) {
logger.debug("Executing callback on JMS Session: " + session);
}
return action.doInJms(session);
}
catch (JMSException ex) {
throw convertJmsAccessException(ex);
}
finally {
JmsUtils.closeSession(session);
ConnectionFactoryUtils.releaseConnection(con, getConnectionFactory(), startConnection);
}
}
//---------------------------------------------------------------------------------------
// Convenience methods for browsing messages
//-------------------------------------------------------------------------
//---------------------------------------------------------------------------------------
@Override
public <T> T browse(BrowserCallback<T> action) throws JmsException {