spring/spring-framework
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-framework.git synced 2026-09-17 08:24:13 +00:00
Code Issues Projects Releases Wiki Activity

Use correlation id for explicit sendAndReceive response queue

Browse Source 
See gh-36162
See gh-36177
This commit is contained in:
Juergen Hoeller
2026-01-21 14:46:36 +01:00
parent 8fbd147f0b
commit 1c4b20287f
3 changed files with 105 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spring-jms/src/main/java/org/springframework/jms/core/JmsOperations.java
+17 -13
View File
@@ -361,7 +361,7 @@ public interface JmsOperations {
/**
* 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}
* 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
@@ -373,10 +373,10 @@ public interface JmsOperations {
@Nullable 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.
* Send a message and receive the reply from the specified destination.
* <p>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,
@@ -387,10 +387,10 @@ public interface JmsOperations {
@Nullable 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.
* Send a message and receive the reply from the specified destination.
* <p>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
@@ -405,8 +405,10 @@ public interface JmsOperations {
* Send a message to the specified destination and receive the reply from the
* specified response queue.
* <p>The {@link MessageCreator} callback creates the message given a Session,
* and the specified {@code responseQueue} is set in the {@code JMSReplyTO}
* header of the message.
* potentially setting a {@code JMSCorrelationID} for receiving from the given
* response queue; otherwise, the {@code JMSMessageID} is used for correlation.
* The specified {@code responseQueue} is set in the {@code JMSReplyTo} header
* of the message.
* @param destination the destination to send the message to
* @param responseQueue the destination to receive the reply from
* @param messageCreator callback to create a message
@@ -422,8 +424,10 @@ public interface JmsOperations {
* Send a message to the specified destination and receive the reply from the
* specified response queue.
* <p>The {@link MessageCreator} callback creates the message given a Session,
* and the destination with the specified {@code responseQueueName} is set in
* the {@code JMSReplyTO} header of the message.
* potentially setting a {@code JMSCorrelationID} for receiving from the given
* response queue; otherwise, the {@code JMSMessageID} is used for correlation.
* The specified {@code responseQueue} is set in the {@code JMSReplyTo} header
* of the message.
* @param destinationName the name of the destination to send the message to
* (to be resolved to an actual destination by a DestinationResolver)
* @param responseQueueName the name of the destination to receive the reply from
spring-jms/src/main/java/org/springframework/jms/core/JmsTemplate.java
+19 -10
View File
@@ -898,11 +898,6 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
return executeLocal(session -> doSendAndReceive(session, destination, messageCreator), true);
}
@Override
public @Nullable Message sendAndReceive(Destination destination, Destination responseQueue, MessageCreator messageCreator) throws JmsException {
return executeLocal(session -> doSendAndReceive(session, destination, responseQueue, messageCreator), true);
}
@Override
public @Nullable Message sendAndReceive(String destinationName, MessageCreator messageCreator) throws JmsException {
return executeLocal(session -> {
@@ -911,12 +906,17 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
}, true);
}
@Override
public @Nullable Message sendAndReceive(Destination destination, Destination responseQueue, MessageCreator messageCreator) throws JmsException {
return executeLocal(session -> doSendAndReceive(session, destination, responseQueue, messageCreator, true), true);
}
@Override
public @Nullable Message sendAndReceive(String destinationName, String responseQueueName, MessageCreator messageCreator) throws JmsException {
return executeLocal(session -> {
Destination destination = resolveDestinationName(session, destinationName);
Destination responseQueue = resolveDestinationName(session, responseQueueName);
return doSendAndReceive(session, destination, responseQueue, messageCreator);
return doSendAndReceive(session, destination, responseQueue, messageCreator, true);
}, true);
}
@@ -932,7 +932,7 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
TemporaryQueue responseQueue = null;
try {
responseQueue = session.createTemporaryQueue();
return doSendAndReceive(session, destination, responseQueue, messageCreator);
return doSendAndReceive(session, destination, responseQueue, messageCreator, false);
}
finally {
if (responseQueue != null) {
@@ -946,9 +946,10 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
* a reply has been received on the specified {@link Destination responseQueue}.
* <p>Return the response message or {@code null} if no message has been received.
* @throws JMSException if thrown by JMS API methods
* @since 7.0.4
*/
protected @Nullable Message doSendAndReceive(Session session, Destination destination, Destination responseQueue, MessageCreator messageCreator)
throws JMSException {
protected @Nullable Message doSendAndReceive(Session session, Destination destination, Destination responseQueue,
MessageCreator messageCreator, boolean useCorrelationId) throws JMSException {
Assert.notNull(messageCreator, "MessageCreator must not be null");
MessageProducer producer = null;
@@ -956,12 +957,20 @@ public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations
try {
Message requestMessage = messageCreator.createMessage(session);
producer = session.createProducer(destination);
consumer = session.createConsumer(responseQueue);
requestMessage.setJMSReplyTo(responseQueue);
if (logger.isDebugEnabled()) {
logger.debug("Sending created message: " + requestMessage);
}
doSend(producer, requestMessage);
String messageSelector = null;
if (useCorrelationId) {
String correlationId = requestMessage.getJMSCorrelationID();
if (correlationId == null) {
correlationId = requestMessage.getJMSMessageID();
}
messageSelector = "JMSCorrelationID='" + correlationId + "'";
}
consumer = session.createConsumer(responseQueue, messageSelector);
return receiveFromConsumer(consumer, getReceiveTimeout());
}
finally {
spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateTests.java
+69 -6
View File
@@ -648,8 +648,7 @@ class JmsTemplateTests {
given(localSession.createTemporaryQueue()).willReturn(replyDestination);
MessageConsumer messageConsumer = mock();
given(localSession.createConsumer(replyDestination)).willReturn(messageConsumer);
given(localSession.createConsumer(replyDestination, null)).willReturn(messageConsumer);
TextMessage request = mock();
MessageCreator messageCreator = mock();
@@ -697,9 +696,7 @@ class JmsTemplateTests {
doTestSendAndReceiveWithResponseQueue(false, 1000L);
}
private void doTestSendAndReceiveWithResponseQueue(boolean explicitDestination, long timeout)
throws Exception {
private void doTestSendAndReceiveWithResponseQueue(boolean explicitDestination, long timeout) throws Exception {
JmsTemplate template = createTemplate();
template.setConnectionFactory(this.connectionFactory);
template.setReceiveTimeout(timeout);
@@ -715,9 +712,75 @@ class JmsTemplateTests {
given(localSession.createProducer(this.queue)).willReturn(messageProducer);
MessageConsumer messageConsumer = mock();
given(localSession.createConsumer(responseQueue)).willReturn(messageConsumer);
// Default sendAndReceive with responseQueue uses MESSAGE_ID selector
given(localSession.createConsumer(responseQueue, "JMSCorrelationID='ID:test-message-id-12345'"))
.willReturn(messageConsumer);
TextMessage request = mock();
given(request.getJMSMessageID()).willReturn("ID:test-message-id-12345");
MessageCreator messageCreator = mock();
given(messageCreator.createMessage(localSession)).willReturn(request);
TextMessage reply = mock();
if (timeout == JmsTemplate.RECEIVE_TIMEOUT_NO_WAIT) {
given(messageConsumer.receiveNoWait()).willReturn(reply);
}
else if (timeout == JmsTemplate.RECEIVE_TIMEOUT_INDEFINITE_WAIT) {
given(messageConsumer.receive()).willReturn(reply);
}
else {
given(messageConsumer.receive(timeout)).willReturn(reply);
}
Message message;
if (explicitDestination) {
message = template.sendAndReceive(this.queue, responseQueue, messageCreator);
}
else {
message = template.sendAndReceive(destinationName, responseQueueName, messageCreator);
}
// replyTO set on the request
verify(request).setJMSReplyTo(responseQueue);
assertThat(message).as("Reply message not received").isSameAs(reply);
verify(this.connection).start();
verify(this.connection).close();
verify(localSession).close();
verify(messageConsumer).close();
verify(messageProducer).close();
}
@Test
void testSendAndReceiveDestinationWithResponseQueueAndCorrelationIdSelector() throws Exception {
doTestSendAndReceiveWithResponseQueueAndCorrelationId(true, 1000L);
}
@Test
void testSendAndReceiveDestinationNameWithResponseQueueNameAndCorrelationIdSelector() throws Exception {
doTestSendAndReceiveWithResponseQueueAndCorrelationId(false, 1000L);
}
private void doTestSendAndReceiveWithResponseQueueAndCorrelationId(boolean explicitDestination, long timeout) throws Exception {
JmsTemplate template = createTemplate();
template.setConnectionFactory(this.connectionFactory);
template.setReceiveTimeout(timeout);
String destinationName = "testDestination";
String responseQueueName = "responseQueue";
Queue responseQueue = mock();
given(this.jndiContext.lookup(responseQueueName)).willReturn(responseQueue);
Session localSession = getLocalSession();
MessageProducer messageProducer = mock();
given(localSession.createProducer(this.queue)).willReturn(messageProducer);
MessageConsumer messageConsumer = mock();
given(localSession.createConsumer(responseQueue, "JMSCorrelationID='xyz'"))
.willReturn(messageConsumer);
TextMessage request = mock();
given(request.getJMSCorrelationID()).willReturn("xyz");
MessageCreator messageCreator = mock();
given(messageCreator.createMessage(localSession)).willReturn(request);