From 34c938fe4fc0fe3505faf93bbf10449cbad72859 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 18 Jan 2018 14:06:17 -0500 Subject: [PATCH] INT-4379: JMS OG Shutdown reply container on stop JIRA: https://jira.spring.io/browse/INT-4379 - shutdown the container when the gateway is stopped Also, improve test suite - at the end of the tests, hundreds of threads are running, some caused by the above but others because `TaskExecutor`s are not shut down - reduce the number of iterations in the JMS pipeline tests to speed things up - change more tests to extend `ActiveMQMultiContextTests`, to keep a single broker up __cherry-pick to 4.3.x__ # Conflicts: # spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java # spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java # spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java # spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java # spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java # spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java --- .../integration/jms/JmsOutboundGateway.java | 11 +- .../jms/JmsOutboundGatewayTests.java | 41 ++-- .../jms/JmsOutboundInsideChainTests.java | 5 +- .../jms/OutboundGatewayConnectionTests.java | 45 ++--- .../jms/OutboundGatewayFunctionTests.java | 177 ++++++++---------- .../jms/PollableJmsChannelTests.java | 30 ++- .../jms/SubscribableJmsChannelTests.java | 4 +- .../jms/request_reply/PipelineJmsTests.java | 18 +- .../PipelineNamedReplyQueuesJmsTests.java | 18 +- ...eplyScenariosWithTempReplyQueuesTests.java | 41 ++-- 10 files changed, 193 insertions(+), 197 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 35d10c52d9..70d6ea5da8 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-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -156,6 +156,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private volatile long idleReplyContainerTimeout; + private volatile boolean wasStopped; + private ScheduledFuture idleTask; /** @@ -676,6 +678,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp if (this.replyContainer != null) { TaskScheduler taskScheduler = getTaskScheduler(); if (this.idleReplyContainerTimeout <= 0) { + if (this.wasStopped) { + this.replyContainer.initialize(); + this.wasStopped = false; + } this.replyContainer.start(); } else { @@ -695,7 +701,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp public void stop() { synchronized (this.lifeCycleMonitor) { if (this.replyContainer != null) { - this.replyContainer.stop(); + this.replyContainer.shutdown(); + this.wasStopped = true; this.deleteDestinationIfTemporary(this.replyContainer.getDestination()); if (this.reaper != null) { this.reaper.cancel(false); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java index 2df90ad379..64622193c7 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java @@ -28,6 +28,7 @@ import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; @@ -69,13 +70,11 @@ import org.springframework.util.ObjectUtils; * @author Artem Bilan * @since 2.2.4 */ -public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { +public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests { private final Log logger = LogFactory.getLog(this.getClass()); - public JmsOutboundGatewayTests() { - super("org.springframework.integration", "org.springframework.jms", "org.apache"); - } + @Test public void testContainerBeanNameWhenNoGatewayBeanName() { @@ -100,16 +99,11 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { gateway.setUseReplyContainer(true); ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties(); final List errors = new ArrayList(); + ExecutorService exec = Executors.newFixedThreadPool(10); ErrorHandlingTaskExecutor errorHandlingTaskExecutor = - new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() { - - @Override - public void handleError(Throwable t) { - logger.info("Error:", t); - errors.add(t); - throw new RuntimeException(t); - } - + new ErrorHandlingTaskExecutor(exec, t -> { + errors.add(t); + throw new RuntimeException(t); }); replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor); replyContainerProperties.setRecoveryInterval(100L); @@ -173,6 +167,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { } finally { gateway.stop(); + exec.shutdownNow(); } } @@ -192,13 +187,8 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { gateway.setReceiveTimeout(60000); gateway.afterPropertiesSet(); gateway.start(); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - gateway.handleMessage(new GenericMessage("foo")); - } - }); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> gateway.handleMessage(new GenericMessage("foo"))); CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); JmsTemplate template = new JmsTemplate(connectionFactory2); @@ -224,6 +214,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { gateway.stop(); connectionFactory1.destroy(); connectionFactory2.destroy(); + exec.shutdownNow(); } @Test @@ -243,13 +234,8 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { gateway.setCorrelationKey("JMSCorrelationID"); gateway.afterPropertiesSet(); gateway.start(); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - gateway.handleMessage(new GenericMessage("foo")); - } - }); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> gateway.handleMessage(new GenericMessage("foo"))); CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); JmsTemplate template = new JmsTemplate(connectionFactory2); @@ -276,6 +262,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport { gateway.stop(); connectionFactory1.destroy(); connectionFactory2.destroy(); + exec.shutdownNow(); } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java index 665342aba4..51d0882251 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -36,11 +36,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * //INT-2275 * * @author Artem Bilan + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration @DirtiesContext -public class JmsOutboundInsideChainTests { +public class JmsOutboundInsideChainTests extends ActiveMQMultiContextTests { @Autowired private MessageChannel outboundChainChannel; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java index b27ff85829..d04b001aa3 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -22,6 +22,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -37,12 +38,13 @@ import org.apache.activemq.broker.BrokerService; import org.apache.activemq.command.ActiveMQQueue; import org.junit.Ignore; import org.junit.Test; + import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.messaging.support.GenericMessage; import org.springframework.jms.connection.CachingConnectionFactory; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -57,14 +59,15 @@ public class OutboundGatewayConnectionTests { private Destination replyQueue1 = new ActiveMQQueue("reply1"); - @Test @Ignore // need a more reliable stop/start for AMQ + @Test + @Ignore // need a more reliable stop/start for AMQ public void testContainerWithDestBrokenConnection() throws Exception { BeanFactory beanFactory = mock(BeanFactory.class); when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); BrokerService broker = new BrokerService(); @@ -82,15 +85,14 @@ public class OutboundGatewayConnectionTests { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -116,15 +118,13 @@ public class OutboundGatewayConnectionTests { final CountDownLatch latch3 = new CountDownLatch(1); final CountDownLatch latch4 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - public void run() { - latch3.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch4.countDown(); - } + exec.execute(() -> { + latch3.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch4.countDown(); } }); assertTrue(latch3.await(10, TimeUnit.SECONDS)); @@ -147,6 +147,7 @@ public class OutboundGatewayConnectionTests { broker.stop(); scheduler.destroy(); + exec.shutdownNow(); } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java index 802533b43d..1557ce34d6 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java @@ -23,6 +23,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -40,7 +41,6 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.integration.test.support.LogAdjustingTestSupport; import org.springframework.integration.test.util.TestUtils; import org.springframework.jms.JmsException; import org.springframework.jms.connection.CachingConnectionFactory; @@ -56,7 +56,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; * @since 2.2 * */ -public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { +public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests { private static Destination requestQueue1 = new ActiveMQQueue("request1"); @@ -85,7 +85,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); ConnectionFactory connectionFactory = getConnectionFactory(); @@ -99,16 +99,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -130,6 +128,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); scheduler.destroy(); + exec.shutdown(); } @Test @@ -139,7 +138,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); gateway.setConnectionFactory(getConnectionFactory()); @@ -151,16 +150,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -183,6 +180,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -192,7 +190,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); gateway.setConnectionFactory(getConnectionFactory()); @@ -205,16 +203,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -236,6 +232,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -245,7 +242,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); gateway.setConnectionFactory(getConnectionFactory()); @@ -257,16 +254,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -289,6 +284,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -298,7 +294,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); gateway.setConnectionFactory(getConnectionFactory()); @@ -311,16 +307,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -342,6 +336,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -352,7 +347,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); gateway.setConnectionFactory(getConnectionFactory()); @@ -363,16 +358,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - latch1.countDown(); - try { - reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); - } - finally { - latch2.countDown(); - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo"))); + } + finally { + latch2.countDown(); } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); @@ -395,6 +388,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -404,7 +398,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.initialize(); when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) - .thenReturn(scheduler); + .thenReturn(scheduler); final JmsOutboundGateway gateway = new JmsOutboundGateway(); gateway.setBeanFactory(beanFactory); gateway.setConnectionFactory(getConnectionFactory()); @@ -417,33 +411,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.setReceiveTimeout(20000); gateway.afterPropertiesSet(); gateway.start(); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - JmsTemplate template = new JmsTemplate(); - template.setConnectionFactory(getConnectionFactory()); - template.setReceiveTimeout(20000); - receiveAndSend(template); - receiveAndSend(template); - } - - private void receiveAndSend(JmsTemplate template) { - javax.jms.Message request = template.receive(requestQueue7); - final javax.jms.Message jmsReply = request; - try { - template.send(request.getJMSReplyTo(), new MessageCreator() { - - @Override - public Message createMessage(Session session) throws JMSException { - return jmsReply; - } - }); - } - catch (JmsException e) { - } - catch (JMSException e) { - } - } + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getConnectionFactory()); + template.setReceiveTimeout(20000); + receiveAndSend(template); + receiveAndSend(template); }); assertNotNull(gateway.handleRequestMessage(new GenericMessage("foo"))); @@ -460,6 +434,17 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { gateway.stop(); assertFalse(container.isRunning()); scheduler.destroy(); + exec.shutdownNow(); + } + + private void receiveAndSend(JmsTemplate template) { + javax.jms.Message request = template.receive(requestQueue7); + final javax.jms.Message jmsReply = request; + try { + template.send(request.getJMSReplyTo(), session -> jmsReply); + } + catch (JmsException | JMSException e) { + } } private ConnectionFactory getConnectionFactory() { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java index 939c54a749..6842faab43 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -28,9 +28,11 @@ import static org.mockito.Mockito.verify; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; + import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Session; @@ -61,7 +63,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Gunnar Hillert * @author Artem Bilan */ -public class PollableJmsChannelTests { +public class PollableJmsChannelTests extends ActiveMQMultiContextTests { private ActiveMQConnectionFactory connectionFactory; @@ -190,14 +192,10 @@ public class PollableJmsChannelTests { assertTrue(sent1); final AtomicReference message = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - message.set(receiver.receive(queue)); - latch1.countDown(); - } - + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + message.set(receiver.receive(queue)); + latch1.countDown(); }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); assertNotNull(message.get()); @@ -208,20 +206,16 @@ public class PollableJmsChannelTests { final CountDownLatch latch2 = new CountDownLatch(1); boolean sent2 = channel.send(MessageBuilder.withPayload("foo").setPriority(6).build()); assertTrue(sent2); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - message.set(receiver.receive(queue)); - latch2.countDown(); - } - + exec.execute(() -> { + message.set(receiver.receive(queue)); + latch2.countDown(); }); assertTrue(latch2.await(10, TimeUnit.SECONDS)); assertNotNull(message.get()); assertEquals(6, message.get().getJMSPriority()); assertTrue(message.get().getJMSExpiration() <= System.currentTimeMillis() + ttl); assertTrue(message.get().toString().contains("persistent = false")); + exec.shutdownNow(); } @Test diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java index 206a28384c..a0bf6a4c61 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -70,7 +70,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Artem Bilan * @since 2.0 */ -public class SubscribableJmsChannelTests { +public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests { private static final int TIMEOUT = 30000; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java index e74a28e9b5..99369ece49 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -20,7 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; @@ -28,6 +28,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.log4j.Level; import org.apache.log4j.LogManager; +import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -39,6 +40,7 @@ import org.springframework.integration.jms.ActiveMQMultiContextTests; import org.springframework.integration.jms.config.ActiveMqTestUtils; import org.springframework.integration.test.support.LongRunningIntegrationTest; import org.springframework.messaging.support.GenericMessage; + /** * @author Oleg Zhurakousky * @author Gary Russell @@ -46,7 +48,7 @@ import org.springframework.messaging.support.GenericMessage; */ public class PipelineJmsTests extends ActiveMQMultiContextTests { - private final Executor executor = Executors.newFixedThreadPool(30); + private final ExecutorService executor = Executors.newFixedThreadPool(30); private static final Log logger = LogFactory.getLog(PipelineJmsTests.class); @@ -58,7 +60,12 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests { LogManager.getLogger(getClass()).setLevel(Level.INFO); } - int requests = 50; + @After + public void tearDown() { + this.executor.shutdownNow(); + } + + int requests = 5; /** * jms:out -> jms:in -> randomTimeoutProcess -> @@ -162,6 +169,7 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests { for (int i = 0; i < requests; i++) { final int y = i; executor.execute(new Runnable() { + @Override public void run() { try { @@ -189,7 +197,7 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests { logger.info("Failure: " + failureCounter.get()); // technically all we care that its > 0, // but reality of this test it has to be something more then 0 - assertTrue(successCounter.get() > 10); + assertTrue(successCounter.get() > 1); assertEquals(0, failureCounter.get()); assertEquals(requests, successCounter.get() + timeoutCounter.get()); context.close(); 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 c72dab9743..8b40462768 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -21,7 +21,7 @@ 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.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -31,6 +31,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.log4j.Level; import org.apache.log4j.LogManager; +import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -43,6 +44,7 @@ import org.springframework.integration.jms.config.ActiveMqTestUtils; import org.springframework.integration.test.support.LongRunningIntegrationTest; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; + /** * @author Oleg Zhurakousky * @author Gary Russell @@ -51,7 +53,7 @@ import org.springframework.messaging.support.GenericMessage; */ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests { - private final Executor executor = Executors.newFixedThreadPool(30); + private final ExecutorService executor = Executors.newFixedThreadPool(30); private static final Log logger = LogFactory.getLog(PipelineJmsTests.class); @@ -63,7 +65,12 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests LogManager.getLogger(getClass()).setLevel(Level.INFO); } - int requests = 50; + @After + public void tearDown() { + this.executor.shutdownNow(); + } + + int requests = 5; /** * jms:out(reply-destination-name="pipeline01-01") -> jms:in -> randomTimeoutProcess -> @@ -175,6 +182,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests for (int i = 1000000; i < 1000000 + requests * 100000; i += 100000) { final int y = i; executor.execute(new Runnable() { + @Override public void run() { try { @@ -197,7 +205,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests assertTrue(latch.await(120, TimeUnit.SECONDS)); // technically all we care that its > 0, // but reality of this test it has to be something more then 0 - assertTrue(successCounter.get() > 10); + assertTrue(successCounter.get() > 1); assertEquals(0, failureCounter.get()); assertEquals(requests, successCounter.get() + timeoutCounter.get()); return timeoutCounter.get(); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java index 65508757b3..bacdb791d9 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -19,11 +19,9 @@ package org.springframework.integration.jms.request_reply; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; -import java.util.ArrayList; -import java.util.List; import java.util.Random; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; @@ -56,6 +54,7 @@ import org.springframework.jms.listener.SessionAwareMessageListener; import org.springframework.jms.support.converter.SimpleMessageConverter; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.support.GenericMessage; + /** * @author Oleg Zhurakousky * @author Gary Russell @@ -73,7 +72,8 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti public void messageCorrelationBasedOnRequestMessageId() throws Exception { ActiveMqTestUtils.prepare(); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass()); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass()); RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class); final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); @@ -187,7 +187,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti BrokerService broker = new BrokerService(); broker.setPersistent(false); broker.setUseJmx(false); - broker.setTransportConnectorURIs(new String[]{"tcp://localhost:61623"}); + broker.setTransportConnectorURIs(new String[] { "tcp://localhost:61623" }); broker.setDeleteAllMessagesOnStartup(true); broker.start(); @@ -223,7 +223,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("mult-producer-and-consumers-temp-reply.xml", this.getClass()); final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); - Executor executor = Executors.newFixedThreadPool(10); + ExecutorService executor = Executors.newFixedThreadPool(10); final int testNumbers = 100; final CountDownLatch latch = new CountDownLatch(testNumbers); final AtomicInteger failures = new AtomicInteger(); @@ -232,6 +232,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti for (int i = 0; i < testNumbers; i++) { final int y = i; executor.execute(new Runnable() { + @Override public void run() { try { @@ -266,6 +267,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti assertEquals(0, failures.get()); assertEquals(0, timeouts.get()); context.close(); + executor.shutdownNow(); } private void print(AtomicInteger failures, AtomicInteger timeouts, AtomicInteger missmatches, long echangesProcessed) { @@ -273,20 +275,10 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti logger.info(echangesProcessed + " exchanges processed"); logger.info("Failures: " + failures.get()); logger.info("Timeouts: " + timeouts.get()); - logger.info("Missmatches: " + missmatches.get()); + logger.info("Mismatches: " + missmatches.get()); logger.info("============================"); } - public static class MyRandomlySlowService { - Random random = new Random(); - List list = new ArrayList(); - public String secho(String value) throws Exception { - int i = random.nextInt(2000); - Thread.sleep(i); - return value; - } - } - private Object extractPayload(Message jmsMessage) { try { return converter.fromMessage(jmsMessage); @@ -297,4 +289,17 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti } return null; } + + public static class MyRandomlySlowService { + + Random random = new Random(); + + public String echo(String value) throws Exception { + int i = random.nextInt(2000); + Thread.sleep(i); + return value; + } + + } + }