From cda5a07aadf6494596674b2e1d8b09f1f09d5dd3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 20 May 2014 14:43:31 +0300 Subject: [PATCH] INT-3405 JMSCorrelationID* for JMSOutboundGateway JIRA: https://jira.spring.io/browse/INT-3405 Disallow `JMSCorrelationID*` with reply-container Polishing; Docs, Test --- .../integration/jms/JmsOutboundGateway.java | 55 +++++++++++++------ ...nariosWithCorrelationKeyProvidedTests.java | 23 +++++++- .../explicit-correlation-key.xml | 31 ++++++++++- src/reference/docbook/jms.xml | 13 +++++ 4 files changed, 102 insertions(+), 20 deletions(-) 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. + +