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 1140abd2f7..380e711abb 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -17,7 +17,6 @@ package org.springframework.integration.jms; import java.util.Date; -import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; @@ -42,6 +41,9 @@ import javax.jms.TemporaryQueue; import javax.jms.TemporaryTopic; import javax.jms.Topic; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.context.SmartLifecycle; import org.springframework.expression.Expression; import org.springframework.integration.Message; @@ -133,7 +135,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private final String gatewayCorrelation = UUID.randomUUID().toString(); private final Map> replies = - new HashMap>(); + new ConcurrentHashMap>(); private final ConcurrentHashMap earlyOrLateReplies = new ConcurrentHashMap(); @@ -236,7 +238,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp * necessary when providing a destination name for a Topic rather than * a destination reference. * - * @param requestPubSubDomain true if the request destination is a Topic + * @param requestPubSubDomain true if the request destination is a Topic. */ public void setRequestPubSubDomain(boolean requestPubSubDomain) { this.requestPubSubDomain = requestPubSubDomain; @@ -247,7 +249,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp * necessary when providing a destination name for a Topic rather than * a destination reference. * - * @param replyPubSubDomain true if the reply destination is a Topic + * @param replyPubSubDomain true if the reply destination is a Topic. */ public void setReplyPubSubDomain(boolean replyPubSubDomain) { this.replyPubSubDomain = replyPubSubDomain; @@ -256,6 +258,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Set the max timeout value for the MessageConsumer's receive call when * waiting for a reply. The default value is 5 seconds. + * + * @param receiveTimeout The receive timeout. */ public void setReceiveTimeout(long receiveTimeout) { this.receiveTimeout = receiveTimeout; @@ -277,6 +281,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Specify the JMS priority to use when sending request Messages. * The value should be within the range of 0-9. + * + * @param priority The priority. */ public void setPriority(int priority) { this.priority = priority; @@ -285,6 +291,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Specify the timeToLive for each sent Message. * The default value indicates no expiration. + * + * @param timeToLive The time to live. */ public void setTimeToLive(long timeToLive) { this.timeToLive = timeToLive; @@ -293,6 +301,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Specify whether explicit QoS settings are enabled * (deliveryMode, priority, and timeToLive). + * + * @param explicitQosEnabled true to enable explicit QoS. */ public void setExplicitQosEnabled(boolean explicitQosEnabled) { this.explicitQosEnabled = explicitQosEnabled; @@ -303,11 +313,19 @@ 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) { this.correlationKey = correlationKey; @@ -319,6 +337,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp * the JMS reply Messages back into Spring Integration Messages. *

* The default is {@link SimpleMessageConverter}. + * + * @param messageConverter The message converter. */ public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "'messageConverter' must not be null"); @@ -328,6 +348,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Provide a {@link JmsHeaderMapper} implementation for mapping the * Spring Integration Message Headers to/from JMS Message properties. + * + * @param headerMapper The header mapper. */ public void setHeaderMapper(JmsHeaderMapper headerMapper) { this.headerMapper = headerMapper; @@ -344,7 +366,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp *
* Default is 'true' * - * @param extractRequestPayload + * @param extractRequestPayload true to extract the request payload. */ public void setExtractRequestPayload(boolean extractRequestPayload) { this.extractRequestPayload = extractRequestPayload; @@ -357,7 +379,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp * Otherwise, the entire JMS Message will become the payload of the * Spring Integration Message. * - * @param extractReplyPayload + * @param extractReplyPayload true to extract the reply payload. */ public void setExtractReplyPayload(boolean extractReplyPayload) { this.extractReplyPayload = extractReplyPayload; @@ -366,6 +388,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Specify the Spring Integration reply channel. If this property is not * set the gateway will check for a 'replyChannel' header on the request. + * + * @param replyChannel The reply channel. */ public void setReplyChannel(MessageChannel replyChannel) { this.setOutputChannel(replyChannel); @@ -454,10 +478,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp session, replyDestinationName, this.replyPubSubDomain); } + @Override public int getPhase() { return Integer.MAX_VALUE; } + @Override public boolean isAutoStartup() { return this.autoStartup; } @@ -576,6 +602,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp } } + @Override public void start() { synchronized (this.lifeCycleMonitor) { if (!this.active) { @@ -590,6 +617,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp } } + @Override public void stop() { synchronized (this.lifeCycleMonitor) { if (this.replyContainer != null) { @@ -601,10 +629,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp } } + @Override public boolean isRunning() { return this.active; } + @Override public void stop(Runnable callback) { this.stop(); callback.run(); @@ -963,6 +993,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Create a new JMS Connection for this JMS gateway. + * + * @return The connection. + * @throws JMSException Any JMSException. */ protected Connection createConnection() throws JMSException { return this.connectionFactory.createConnection(); @@ -970,11 +1003,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp /** * Create a new JMS Session using the provided Connection. + * + * @param connection The connection. + * @return The session. + * @throws JMSException Any JMSException. */ protected Session createSession(Connection connection) throws JMSException { return connection.createSession(false, Session.AUTO_ACKNOWLEDGE); } + @Override public void onMessage(javax.jms.Message message) { String correlationId = null; try { @@ -991,6 +1029,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp LinkedBlockingQueue queue = this.replies.get(correlationId); if (queue == null) { if (this.correlationKey != null) { + Log debugLogger = LogFactory.getLog("si.jmsgateway.debug"); + if (debugLogger.isDebugEnabled()) { + Object siMessage = this.messageConverter.fromMessage(message); + debugLogger.debug("No pending reply for " + siMessage + " with correlationId: " + + correlationId + " pending replies: " + this.replies.keySet()); + } throw new RuntimeException("No sender waiting for reply"); } synchronized (this.earlyOrLateReplies) { @@ -1019,7 +1063,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private class GatewayReplyListenerContainer extends DefaultMessageListenerContainer { - private Destination replyDestination; + private volatile Destination replyDestination; @Override protected Destination resolveDestinationName(Session session, String destinationName) throws JMSException { @@ -1123,6 +1167,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private class LateReplyReaper implements Runnable { + @Override public void run() { if (logger.isTraceEnabled()) { logger.trace("Running late reply reaper"); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java index 1277f0bf76..082b5e53cd 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java @@ -18,9 +18,11 @@ package org.springframework.integration.jms.request_reply; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -39,6 +41,8 @@ import org.springframework.integration.jms.ActiveMQMultiContextTests; import org.springframework.integration.jms.config.ActiveMqTestUtils; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.test.support.LongRunningIntegrationTest; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; /** * @author Oleg Zhurakousky * @author Gary Russell diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml index aa0d9ed4d4..5990be1c34 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml @@ -10,6 +10,7 @@ + + + + + + + + @@ -33,6 +42,7 @@ + + + + + - + + + + - + + + + - + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml index 0c2fbd8076..deb9a50c6c 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml @@ -10,14 +10,22 @@ + + + + + + + + @@ -34,20 +43,29 @@ + + + + + - + + + + diff --git a/spring-integration-jms/src/test/resources/log4j.properties b/spring-integration-jms/src/test/resources/log4j.properties index b8ebc43ebe..4eb3e380a2 100644 --- a/spring-integration-jms/src/test/resources/log4j.properties +++ b/spring-integration-jms/src/test/resources/log4j.properties @@ -6,6 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m log4j.category.org.springframework=ERROR -#log4j.category.org.springframework.integration.jms=DEBUG -# log4j.category.org.springframework.integration.jdbc=DEBUG -log4j.category.org.springframework.jms=ERROR +log4j.category.org.springframework.integration=ERROR +log4j.category.org.springframework.integration.jms.request_reply=INFO +log4j.category.si.jmsgateway.debug=DEBUG