diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
index 1c6bc31959..3b723d9fa5 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
@@ -318,12 +318,18 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
* the receiver of the JMS Message would expect to represent the CorrelationID.
* When waiting for the reply Message, a MessageSelector will be configured
* to match this property name and the UUID value that was sent in the request.
- * If this value is NULL (the default) then the reply consumer's MessageSelector
+ * If this value is NULL (the default) then the reply consumer's MessageSelector
* will be expecting the JMSCorrelationID to equal the Message ID of the request.
- * If you want to store the outbound correlation UUID value in the actual
+ *
If you want to store the outbound correlation UUID value in the actual
* JMSCorrelationID property, then set this value to "JMSCorrelationID".
- * However, any other value will be treated as a JMS String Property.
- *
+ *
If you want to use and existing "JMSCorrelationID" from the inbound message
+ * (mapped from 'jms_correlationId'),
+ * you can set this property to "JMSCorrelationID*" with the trailing asterisk.
+ * If the message has a correlation id, it will be used, otherwise a new one will
+ * be set in the 'JMSCorrelationID' header. However, understand that the
+ * gateway has no means to ensure uniqueness and unexpected side effects can
+ * occur if the correlation id is not unique.
+ *
This setting is not allowed if a reply listener is used.
* @param correlationKey The correlation key.
*/
public void setCorrelationKey(String correlationKey) {
@@ -530,6 +536,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
this.useReplyContainer = false;
}
if (this.useReplyContainer) {
+ Assert.state(!"JMSCorrelationID*".equals(this.correlationKey),
+ "Using an existing 'JMSCorrelationID' mapped from the 'requestMessage' ('JMSCorrelationID*') " +
+ "can't be used when using a 'reply-container'");
GatewayReplyListenerContainer container = new GatewayReplyListenerContainer();
setContainerProperties(container);
container.afterPropertiesSet();
@@ -715,13 +724,13 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
Destination requestDestination = this.determineRequestDestination(requestMessage, session);
- /*
- * Remove any existing correlation id that was mapped from the inbound message
- * (it will be restored by normal ARPMH header processing).
- */
- jmsRequest.setJMSCorrelationID(null);
javax.jms.Message reply = null;
if (this.correlationKey == null) {
+ /*
+ * Remove any existing correlation id that was mapped from the inbound message
+ * (it will be restored in the reply by normal ARPMH header processing).
+ */
+ jmsRequest.setJMSCorrelationID(null);
reply = doSendAndReceiveAsyncDefaultCorrelation(requestDestination, jmsRequest, session, priority);
}
else {
@@ -796,17 +805,24 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
MessageConsumer messageConsumer = null;
try {
messageProducer = session.createProducer(requestDestination);
- String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
Assert.state(this.correlationKey != null, "correlationKey must not be null");
String messageSelector = null;
- if (this.correlationKey.equals("JMSCorrelationID")) {
- jmsRequest.setJMSCorrelationID(correlationId);
- messageSelector = "JMSCorrelationID = '" + correlationId + "'";
+ if (!this.correlationKey.equals("JMSCorrelationID*") || jmsRequest.getJMSCorrelationID() == null) {
+ String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
+ if (this.correlationKey.equals("JMSCorrelationID")) {
+ jmsRequest.setJMSCorrelationID(correlationId);
+ messageSelector = "JMSCorrelationID = '" + correlationId + "'";
+ }
+ else {
+ jmsRequest.setStringProperty(this.correlationKey, correlationId);
+ jmsRequest.setJMSCorrelationID(null);
+ messageSelector = this.correlationKey + " = '" + correlationId + "'";
+ }
}
else {
- jmsRequest.setStringProperty(this.correlationKey, correlationId);
- messageSelector = this.correlationKey + " = '" + correlationId + "'";
+ messageSelector = "JMSCorrelationID = '" + jmsRequest.getJMSCorrelationID() + "'";
}
+
messageConsumer = session.createConsumer(replyTo, messageSelector);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
return this.receiveReplyMessage(messageConsumer);
@@ -875,6 +891,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
else {
jmsRequest.setStringProperty(this.correlationKey, correlationId);
+ /*
+ * Remove any existing correlation id that was mapped from the inbound message
+ * (it will be restored in the reply by normal ARPMH header processing).
+ */
+ jmsRequest.setJMSCorrelationID(null);
}
LinkedBlockingQueue replyQueue = new LinkedBlockingQueue(1);
if (logger.isDebugEnabled()) {
@@ -1018,7 +1039,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
if (logger.isTraceEnabled()) {
logger.trace(this.getComponentName() + " Received " + message);
}
- if (this.correlationKey == null || this.correlationKey.equals("JMSCorrelationID")) {
+ if (this.correlationKey == null ||
+ this.correlationKey.equals("JMSCorrelationID") ||
+ this.correlationKey.equals("JMSCorrelationID*")) {
correlationId = message.getJMSCorrelationID();
}
else {
diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java
index 5e6c3e30c1..d393870ca7 100644
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -13,24 +13,30 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.springframework.integration.jms.request_reply;
import static org.junit.Assert.assertEquals;
+import java.util.UUID;
+
import org.junit.Rule;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
+import org.springframework.integration.jms.JmsHeaders;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
+ * @author Artem Bilan
*/
public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends ActiveMQMultiContextTests {
@@ -59,6 +65,21 @@ public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends Active
context.close();
}
+ @Test
+ public void messageCorrelationBasedOnProvidedJMSCorrelationID() throws Exception{
+ ActiveMqTestUtils.prepare();
+
+ ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
+ RequestReplyExchanger gateway = context.getBean("existingCorrelationKeyGatewayB", RequestReplyExchanger.class);
+
+ String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
+ Message result = gateway.exchange(MessageBuilder.withPayload("foo")
+ .setHeader(JmsHeaders.CORRELATION_ID, correlationId)
+ .build());
+ assertEquals(correlationId, result.getHeaders().get("receivedCorrelationId"));
+ context.close();
+ }
+
@Test
public void messageCorrelationBasedCustomCorrelationKeyDelayedReplies() throws Exception{
ActiveMqTestUtils.prepare();
diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml
index 15bdd98043..9e47c354e8 100644
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml
@@ -38,14 +38,39 @@
+ request-destination="explicitCorrelationJmsOutB"
+ correlation-key="JMSCorrelationID"
+ connection-factory="connectionFactory"/>
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
is used, the correlation-key MUST be specified if an explicit reply-destination
is provided.
+
+ Starting with version 4.0.1 this attribute also supports the value
+ JMSCorrelationID*, which means that if the outbound message already has a
+ JMSCorrelationID (mapped from the jms_correlationId) header,
+ it will be used, instead of generating a new one.
+ Note, the JMSCorrelationID* key is not allowed when using a
+ <reply-container/> because the container needs to set up a
+ message selector during initialization.
+
+ You should understand that the gateway has no means to ensure uniqueness and
+ unexpected side effects can occur if the provided correlation id is not unique.
+
+