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 3b723d9fa5..3631d1b998 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-2014 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.IntegrationMessageHeaderAccessor; @@ -134,7 +136,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(); @@ -1051,6 +1053,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) { @@ -1079,7 +1087,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 { 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 b5cd5cdbfa..f2760f6980 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; @@ -37,8 +39,9 @@ import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.gateway.RequestReplyExchanger; import org.springframework.integration.jms.ActiveMQMultiContextTests; import org.springframework.integration.jms.config.ActiveMqTestUtils; -import org.springframework.messaging.support.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 @@ -113,7 +116,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests */ @Test public void testPipeline3a() throws Exception{ - int timeouts = this.test("pipeline-named-queue-03a.xml", 20000); + int timeouts = this.test("pipeline-named-queue-03a.xml", 50000); assertEquals(0, timeouts); } @@ -123,7 +126,8 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests */ @Test public void testPipeline4() throws Exception{ - this.test("pipeline-named-queue-04.xml"); + int timeouts = this.test("pipeline-named-queue-04.xml", 30000); + assertEquals(0, timeouts); } /** @@ -167,19 +171,23 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); final CountDownLatch latch = new CountDownLatch(requests); - for (int i = 0; i < requests; i++) { + for (int i = 1000000; i < 1000000 + requests * 100000; i += 100000) { final int y = i; executor.execute(new Runnable() { + @Override public void run() { try { assertEquals(y + offset, gateway.exchange(new GenericMessage(y)).getPayload()); successCounter.incrementAndGet(); - } catch (MessageTimeoutException e) { + } + catch (MessageTimeoutException e) { timeoutCounter.incrementAndGet(); - } catch (Throwable t) { + } + catch (Throwable t) { t.printStackTrace(); failureCounter.incrementAndGet(); - } finally { + } + finally { latch.countDown(); } } @@ -198,7 +206,22 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests logger.info("Success: " + successCounter.get()); logger.info("Timeout: " + timeoutCounter.get()); logger.info("Failure: " + failureCounter.get()); - context.destroy(); + if (timeoutCounter.get() > 0 && context.containsBean("capture")) { + logger.info(context.getBean(Capture.class).messages); + } + context.close(); } } + + public static class Capture { + + private final BlockingQueue messages = new LinkedBlockingQueue(); + + public Message capture(Message message) { + messages.add("\n[" + Thread.currentThread().getName() + "] " + message); + return message; + } + + } + } 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