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

INT-3405 JMSCorrelationID* for JMSOutboundGateway

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3405

Disallow `JMSCorrelationID*` with reply-container

Polishing; Docs, Test
This commit is contained in:
Artem Bilan
2014-05-20 14:43:31 +03:00
committed by Gary Russell
parent c5500a82fd
commit cda5a07aad
4 changed files with 102 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -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
* <p>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
* <p>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.
*
* <p>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.
* <p>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<javax.jms.Message> replyQueue = new LinkedBlockingQueue<javax.jms.Message>(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 {

23
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java
View File

@@ -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();

31
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml
View File

@@ -38,14 +38,39 @@
</bean>
<int-jms:inbound-gateway request-channel="requestInB"
request-destination="explicitCorrelationJmsOutB"
correlation-key="JMSCorrelationID"
connection-factory="connectionFactory"/>
request-destination="explicitCorrelationJmsOutB"
correlation-key="JMSCorrelationID"
connection-factory="connectionFactory"/>
<int:transformer input-channel="requestInB" expression="payload"/>
<!-- -->
<int:gateway id="existingCorrelationKeyGatewayB" default-request-channel="existingCorrelationInB"/>
<int-jms:outbound-gateway request-channel="existingCorrelationInB"
connection-factory="connectionFactory"
request-destination="existingCorrelationJmsOutB"
correlation-key="JMSCorrelationID*"/>
<bean id="existingCorrelationJmsOutB" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="existingCorrelationJmsOutB"/>
</bean>
<int-jms:inbound-gateway request-channel="requestExistingCorrelationInB"
request-destination="existingCorrelationJmsOutB"
correlation-key="JMSCorrelationID"
connection-factory="connectionFactory"/>
<int:chain input-channel="requestExistingCorrelationInB" >
<int:header-enricher>
<int:header name="receivedCorrelationId" expression="headers['jms_correlationId']"/>
</int:header-enricher>
<int:transformer expression="payload"/>
</int:chain>
<!-- -->
<int:gateway id="explicitCorrelationKeyGatewayC" default-request-channel="explicitCorrelationInC"/>
<int-jms:outbound-gateway id="outGateway" request-channel="explicitCorrelationInC"

13
src/reference/docbook/jms.xml
View File

@@ -374,6 +374,19 @@
</code> is used, the correlation-key MUST be specified if an explicit <code>reply-destination</code>
is provided.
</para>
<para>
Starting with <emphasis>version 4.0.1</emphasis> this attribute also supports the value
<code>JMSCorrelationID*</code>, which means that if the outbound message already has a
<code>JMSCorrelationID</code> (mapped from the <code>jms_correlationId</code>) header,
it will be used, instead of generating a new one.
Note, the <code>JMSCorrelationID*</code> key is not allowed when using a
<code>&lt;reply-container/&gt;</code> because the container needs to set up a
message selector during initialization.
<important>
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.
</important>
</para>
</callout>
<callout arearefs="jog030">
<para>