From 7db92cbdacc49b6f77a80e76cbce0e2a1972e471 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 9 Dec 2010 14:57:03 -0500 Subject: [PATCH] INT-1672 added the 'correlationKey' property to the JmsOutboundGateway as well as a 'correlation-key' attribute for the JMS element --- .../integration/jms/JmsOutboundGateway.java | 163 ++++++++++++++---- .../jms/config/JmsOutboundGatewayParser.java | 1 + .../jms/config/spring-integration-jms-2.0.xsd | 15 ++ 3 files changed, 144 insertions(+), 35 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 2a986108dc..dde982cfeb 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 @@ -16,6 +16,8 @@ package org.springframework.integration.jms; +import java.util.UUID; + import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; @@ -81,6 +83,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper(); + private volatile String correlationKey; + private volatile boolean extractRequestPayload = true; private volatile boolean extractReplyPayload = true; @@ -88,10 +92,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { private volatile boolean initialized; private final Object initializationMonitor = new Object(); - - public String getComponentType(){ - return "jms:outbound-gateway"; - } /** @@ -196,6 +196,21 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { this.explicitQosEnabled = explicitQosEnabled; } + /** + * Provide the name of a JMS property that should hold a generated UUID that + * 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 + * 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 + * JMSCorrelationID property, then set this value to "JMSCorrelationID". + * However, any other value will be treated as a JMS String Property. + */ + public void setCorrelationKey(String correlationKey) { + this.correlationKey = correlationKey; + } + /** * Provide a {@link MessageConverter} strategy to use for converting the * Spring Integration request Message into a JMS Message and for converting @@ -217,13 +232,13 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } /** - * This property describes how JMS Message should be generated from - * Spring Integration (SI) Message. If set to 'true', the base for JMS Message will be - * SI Message's payload, if set to 'false', then the entire SI Message will serve as - * a base for JMS Message creation. - * - * Since JMS Message is created by the MessageConverter, this really manages what - * is sent to a {@link MessageConverter} - the entire SI Message or only its payload. + * This property describes how a JMS Message should be generated from the + * Spring Integration Message. If set to 'true', the body of the JMS Message will be + * created from the Spring Integration Message's payload (via the MessageConverter). + * If set to 'false', then the entire Spring Integration Message will serve as + * the base for JMS Message creation. Since the JMS Message is created by the + * MessageConverter, this really manages what is sent to the {@link MessageConverter}: + * the entire Spring Integration Message or only its payload. *
* Default is 'true' * @@ -234,9 +249,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } /** - * This property describes what to do with JMS Message after reply was received. - * If set to 'true', the base for SI Message will be JMS Reply Message's payload - * otherwise the entire JMS Message will become a payload of SI Message. + * This property describes what to do with a JMS reply Message. + * If set to 'true', the payload of the Spring Integration Message will be + * created from the JMS Reply Message's body (via MessageConverter). + * Otherwise, the entire JMS Message will become the payload of the + * Spring Integration Message. * * @param extractReplyPayload */ @@ -252,6 +269,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { this.setOutputChannel(replyChannel); } + public String getComponentType() { + return "jms:outbound-gateway"; + } + private Destination getRequestDestination(Session session) throws JMSException { if (this.requestDestination != null) { return this.requestDestination; @@ -317,52 +338,124 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { private javax.jms.Message sendAndReceive(Message requestMessage) throws JMSException { Connection connection = createConnection(); Session session = null; - MessageProducer messageProducer = null; - MessageConsumer messageConsumer = null; Destination replyTo = null; try { session = createSession(connection); + // convert to JMS Message Object objectToSend = requestMessage; - if (this.extractRequestPayload){ + if (this.extractRequestPayload) { objectToSend = requestMessage.getPayload(); } javax.jms.Message jmsRequest = this.messageConverter.toMessage(objectToSend, session); + // map headers headerMapper.fromHeaders(requestMessage.getHeaders(), jmsRequest); - // create JMS Producer and send - messageProducer = session.createProducer(this.getRequestDestination(session)); - + // TODO: support a JmsReplyTo header in the SI Message? replyTo = this.getReplyDestination(session); jmsRequest.setJMSReplyTo(replyTo); connection.start(); - if (this.explicitQosEnabled) { - messageProducer.send(jmsRequest, - this.deliveryMode, this.priority, this.timeToLive); + + javax.jms.Message replyMessage = null; + if (this.correlationKey != null) { + replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(jmsRequest, replyTo, session); + } + else if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) { + replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(jmsRequest, replyTo, session); } else { - messageProducer.send(jmsRequest); + replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(jmsRequest, replyTo, session); } - if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) { - messageConsumer = session.createConsumer(replyTo); - } - else { - String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''"); - String messageSelector = "JMSCorrelationID = '" + messageId + "'"; - messageConsumer = session.createConsumer(replyTo, messageSelector); - } - return (this.receiveTimeout >= 0) ? messageConsumer.receive(receiveTimeout) : messageConsumer.receive(); + return replyMessage; } finally { - JmsUtils.closeMessageProducer(messageProducer); - JmsUtils.closeMessageConsumer(messageConsumer); JmsUtils.closeSession(session); this.deleteDestinationIfTemporary(replyTo); ConnectionFactoryUtils.releaseConnection(connection, this.connectionFactory, true); } } + /** + * Creates the MessageConsumer before sending the request Message since we are generating our own correlationId value for the MessageSelector. + */ + private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(javax.jms.Message jmsRequest, Destination replyTo, Session session) throws JMSException { + MessageProducer messageProducer = null; + MessageConsumer messageConsumer = null; + try { + messageProducer = session.createProducer(this.getRequestDestination(session)); + 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 + "'"; + } + else { + jmsRequest.setStringProperty(this.correlationKey, correlationId); + messageSelector = this.correlationKey + " = '" + correlationId + "'"; + } + messageConsumer = session.createConsumer(replyTo, messageSelector); + this.sendRequestMessage(jmsRequest, messageProducer); + return this.receiveReplyMessage(messageConsumer); + } + finally { + JmsUtils.closeMessageProducer(messageProducer); + JmsUtils.closeMessageConsumer(messageConsumer); + } + } + + /** + * Creates the MessageConsumer before sending the request Message since we do not need any correlation. + */ + private javax.jms.Message doSendAndReceiveWithTemporaryReplyToDestination(javax.jms.Message jmsRequest, Destination replyTo, Session session) throws JMSException { + MessageProducer messageProducer = null; + MessageConsumer messageConsumer = null; + try { + messageProducer = session.createProducer(this.getRequestDestination(session)); + messageConsumer = session.createConsumer(replyTo); + this.sendRequestMessage(jmsRequest, messageProducer); + return this.receiveReplyMessage(messageConsumer); + } + finally { + JmsUtils.closeMessageProducer(messageProducer); + JmsUtils.closeMessageConsumer(messageConsumer); + } + } + + /** + * Creates the MessageConsumer after sending the request Message since we need the MessageID for correlation with a MessageSelector. + */ + private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(javax.jms.Message jmsRequest, Destination replyTo, Session session) throws JMSException { + MessageProducer messageProducer = null; + MessageConsumer messageConsumer = null; + try { + messageProducer = session.createProducer(this.getRequestDestination(session)); + this.sendRequestMessage(jmsRequest, messageProducer); + String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''"); + String messageSelector = "JMSCorrelationID = '" + messageId + "'"; + messageConsumer = session.createConsumer(replyTo, messageSelector); + return this.receiveReplyMessage(messageConsumer); + } + finally { + JmsUtils.closeMessageProducer(messageProducer); + JmsUtils.closeMessageConsumer(messageConsumer); + } + } + + private void sendRequestMessage(javax.jms.Message jmsRequest, MessageProducer messageProducer) throws JMSException { + if (this.explicitQosEnabled) { + messageProducer.send(jmsRequest, this.deliveryMode, this.priority, this.timeToLive); + } + else { + messageProducer.send(jmsRequest); + } + } + + private javax.jms.Message receiveReplyMessage(MessageConsumer messageConsumer) throws JMSException { + return (this.receiveTimeout >= 0) ? messageConsumer.receive(receiveTimeout) : messageConsumer.receive(); + } + /** * Deletes either a {@link TemporaryQueue} or {@link TemporaryTopic}. * Ignores any other {@link Destination} type and also ignores any diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java index f276be80c0..563e0edda3 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java @@ -56,6 +56,7 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-destination"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-destination-name"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "correlation-key"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "header-mapper"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver"); diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd index 5c21dfeb1f..b33751d486 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd @@ -656,6 +656,21 @@ + + + + +