From 1c4b20287f1d8f635867fb58c576849f9313e907 Mon Sep 17 00:00:00 2001 From: Juergen Hoeller Date: Wed, 21 Jan 2026 14:46:36 +0100 Subject: [PATCH] Use correlation id for explicit sendAndReceive response queue See gh-36162 See gh-36177 --- .../jms/core/JmsOperations.java | 30 ++++---- .../springframework/jms/core/JmsTemplate.java | 29 ++++--- .../jms/core/JmsTemplateTests.java | 75 +++++++++++++++++-- 3 files changed, 105 insertions(+), 29 deletions(-) diff --git a/spring-jms/src/main/java/org/springframework/jms/core/JmsOperations.java b/spring-jms/src/main/java/org/springframework/jms/core/JmsOperations.java index 3561bab0183..3319b04555a 100644 --- a/spring-jms/src/main/java/org/springframework/jms/core/JmsOperations.java +++ b/spring-jms/src/main/java/org/springframework/jms/core/JmsOperations.java @@ -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. *

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. + *

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. + *

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. *

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. *

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 diff --git a/spring-jms/src/main/java/org/springframework/jms/core/JmsTemplate.java b/spring-jms/src/main/java/org/springframework/jms/core/JmsTemplate.java index 63e9d18df04..0cf8840dec0 100644 --- a/spring-jms/src/main/java/org/springframework/jms/core/JmsTemplate.java +++ b/spring-jms/src/main/java/org/springframework/jms/core/JmsTemplate.java @@ -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}. *

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 { diff --git a/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateTests.java b/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateTests.java index 5094948582b..6390e9d76a2 100644 --- a/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateTests.java @@ -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);