From 170292acf3bcf1cbb76107ccd594399b206b8b5a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 22 Jan 2018 16:17:20 -0500 Subject: [PATCH] Shutdown executors in TCP/IP tests --- .../ip/tcp/TcpInboundGatewayTests.java | 55 ++++++------ .../ip/tcp/TcpOutboundGatewayTests.java | 49 ++++++----- .../tcp/TcpReceivingChannelAdapterTests.java | 41 +++++---- .../ip/tcp/TcpSendingMessageHandlerTests.java | 69 +++++++-------- .../CachingClientConnectionFactoryTests.java | 5 +- .../tcp/connection/ConnectionEventTests.java | 10 ++- .../ConnectionFactoryShutDownTests.java | 10 ++- .../connection/ConnectionFactoryTests.java | 5 +- .../FailoverClientConnectionFactoryTests.java | 7 +- .../tcp/connection/TcpNioConnectionTests.java | 83 ++++++++++-------- .../TcpNioConnectionWriteTests.java | 56 ++++++++++--- .../tcp/serializer/DeserializationTests.java | 7 +- ...ramPacketMulticastSendingHandlerTests.java | 6 +- .../DatagramPacketSendingHandlerTests.java | 84 ++++++++++--------- .../ip/udp/UdpChannelAdapterTests.java | 30 +++---- 15 files changed, 298 insertions(+), 219 deletions(-) diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index 89dc91cddb..a39b2a18d7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.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,7 +28,6 @@ import java.net.Socket; import java.util.HashSet; import java.util.Set; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -39,6 +38,7 @@ import javax.net.SocketFactory; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.ServiceActivatingHandler; @@ -56,6 +56,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class TcpInboundGatewayTests { @@ -119,30 +121,31 @@ public class TcpInboundGatewayTests { final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { - try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10); - port.set(server.getLocalPort()); - latch1.countDown(); - Socket socket = server.accept(); - socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes()); - byte[] bytes = new byte[12]; - readFully(socket.getInputStream(), bytes); - assertEquals("Echo:Test1\r\n", new String(bytes)); - readFully(socket.getInputStream(), bytes); - assertEquals("Echo:Test2\r\n", new String(bytes)); - latch2.await(); - socket.close(); - server.close(); - done.set(true); - latch3.countDown(); - } - catch (Exception e) { - if (!done.get()) { - e.printStackTrace(); - } - } - }); + new SimpleAsyncTaskExecutor() + .execute(() -> { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10); + port.set(server.getLocalPort()); + latch1.countDown(); + Socket socket = server.accept(); + socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes()); + byte[] bytes = new byte[12]; + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); + latch2.await(); + socket.close(); + server.close(); + done.set(true); + latch3.countDown(); + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port.get()); ccf.setSingleUse(false); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index ccbc647b2b..d201170ec3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -43,7 +43,6 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; -import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -61,6 +60,8 @@ import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; @@ -90,6 +91,8 @@ public class TcpOutboundGatewayTests { private static final Log logger = LogFactory.getLog(TcpOutboundGatewayTests.class); + private AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + @ClassRule public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest(); @@ -101,13 +104,13 @@ public class TcpOutboundGatewayTests { public void testGoodNetSingle() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - final AtomicReference serverSocket = new AtomicReference(); - Executors.newSingleThreadExecutor().execute(() -> { + final AtomicReference serverSocket = new AtomicReference<>(); + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100); serverSocket.set(server); latch.countDown(); - List sockets = new ArrayList(); + List sockets = new ArrayList<>(); int i = 0; while (true) { Socket socket = server.accept(); @@ -158,6 +161,7 @@ public class TcpOutboundGatewayTests { assertTrue(replies.remove("Reply" + i)); } done.set(true); + ccf.stop(); serverSocket.get().close(); } @@ -165,8 +169,8 @@ public class TcpOutboundGatewayTests { public void testGoodNetMultiplex() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - final AtomicReference serverSocket = new AtomicReference(); - Executors.newSingleThreadExecutor().execute(() -> { + final AtomicReference serverSocket = new AtomicReference<>(); + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10); serverSocket.set(server); @@ -213,6 +217,7 @@ public class TcpOutboundGatewayTests { } done.set(true); gateway.stop(); + ccf.stop(); serverSocket.get().close(); } @@ -220,8 +225,8 @@ public class TcpOutboundGatewayTests { public void testGoodNetTimeout() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - final AtomicReference serverSocket = new AtomicReference(); - Executors.newSingleThreadExecutor().execute(() -> { + final AtomicReference serverSocket = new AtomicReference<>(); + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -260,12 +265,12 @@ public class TcpOutboundGatewayTests { Future[] results = (Future[]) new Future[2]; for (int i = 0; i < 2; i++) { final int j = i; - results[j] = (Executors.newSingleThreadExecutor().submit(() -> { + results[j] = (this.executor.submit(() -> { gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); return 0; })); } - Set replies = new HashSet(); + Set replies = new HashSet<>(); int timeouts = 0; for (int i = 0; i < 2; i++) { try { @@ -294,6 +299,7 @@ public class TcpOutboundGatewayTests { } done.set(true); gateway.stop(); + ccf.stop(); serverSocket.get().close(); } @@ -344,7 +350,7 @@ public class TcpOutboundGatewayTests { final AtomicReference lastReceived = new AtomicReference(); final CountDownLatch serverLatch = new CountDownLatch(2); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { latch.countDown(); int i = 0; @@ -398,7 +404,7 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 2; i++) { final int j = i; - results[j] = (Executors.newSingleThreadExecutor().submit(() -> { + results[j] = (this.executor.submit(() -> { gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); return j; })); @@ -433,6 +439,7 @@ public class TcpOutboundGatewayTests { done.set(true); assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()); gateway.stop(); + ccf.stop(); } @Test @@ -442,7 +449,7 @@ public class TcpOutboundGatewayTests { final AtomicBoolean done = new AtomicBoolean(); final CountDownLatch serverLatch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -512,17 +519,18 @@ public class TcpOutboundGatewayTests { done.set(true); gateway.stop(); verify(mockConn1).send(Mockito.any(Message.class)); + factory2.stop(); serverSocket.get().close(); } @Test public void testFailoverCached() throws Exception { - final AtomicReference serverSocket = new AtomicReference(); + final AtomicReference serverSocket = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final CountDownLatch serverLatch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -595,6 +603,7 @@ public class TcpOutboundGatewayTests { done.set(true); gateway.stop(); verify(mockConn1).send(Mockito.any(Message.class)); + factory2.stop(); serverSocket.get().close(); } @@ -667,11 +676,11 @@ public class TcpOutboundGatewayTests { final ServerSocket server) throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - final AtomicReference lastReceived = new AtomicReference(); + final AtomicReference lastReceived = new AtomicReference<>(); final CountDownLatch serverLatch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { - List sockets = new ArrayList(); + this.executor.execute(() -> { + List sockets = new ArrayList<>(); try { latch.countDown(); while (!done.get()) { @@ -793,8 +802,8 @@ public class TcpOutboundGatewayTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { - List sockets = new ArrayList(); + this.executor.execute(() -> { + List sockets = new ArrayList<>(); try { latch.countDown(); while (!done.get()) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 642a6777cf..890908f00a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 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. @@ -33,7 +33,6 @@ import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -46,6 +45,7 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.ServiceActivatingHandler; @@ -64,6 +64,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell + * @author Artem Bilan */ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTests { @@ -97,27 +98,24 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNetClientMode() throws Exception { - final AtomicReference serverSocket = new AtomicReference(); + final AtomicReference serverSocket = new AtomicReference<>(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override - public void run() { - try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10); - serverSocket.set(server); - latch1.countDown(); - Socket socket = server.accept(); - socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes()); - latch2.await(); - socket.close(); - server.close(); - } - catch (Exception e) { - if (!done.get()) { - e.printStackTrace(); - } + new SimpleAsyncTaskExecutor().execute(() -> { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10); + serverSocket.set(server); + latch1.countDown(); + Socket socket = server.accept(); + socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes()); + latch2.await(); + socket.close(); + server.close(); + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); } } }); @@ -416,7 +414,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe handler.setConnectionFactory(scf); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); - Executor te = Executors.newCachedThreadPool(); + Executor te = new SimpleAsyncTaskExecutor(); scf.setTaskExecutor(te); scf.start(); QueueChannel channel = new QueueChannel(); @@ -650,6 +648,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe } private class FailingService { + @SuppressWarnings("unused") public String serviceMethod(byte[] bytes) { throw new RuntimeException("Failed"); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 2bf21dc774..be9dc18439 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.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. @@ -35,8 +35,6 @@ import java.util.List; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -54,6 +52,8 @@ import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; @@ -79,12 +79,14 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell * @author Artem Bilan + * * @since 2.0 */ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTests { private static final Log logger = LogFactory.getLog(TcpSendingMessageHandlerTests.class); + private AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { @@ -97,7 +99,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -150,7 +152,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -207,6 +209,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest handler.start(); handler.stop(); adapter.stop(); + ccf.stop(); serverSocket.get().close(); } @@ -215,7 +218,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -271,7 +274,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -324,7 +327,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -377,10 +380,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest @Test public void testNetLength() throws Exception { - final AtomicReference serverSocket = new AtomicReference(); + final AtomicReference serverSocket = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -436,7 +439,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -495,7 +498,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -544,10 +547,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest @Test public void testNioSerial() throws Exception { - final AtomicReference serverSocket = new AtomicReference(); + final AtomicReference serverSocket = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -598,12 +601,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } @Test - public void testNetSingleUseNoInbound() throws Exception { + public void testNetSingleUseNoInbound() throws Exception { final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -645,12 +648,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } @Test - public void testNioSingleUseNoInbound() throws Exception { + public void testNioSingleUseNoInbound() throws Exception { final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -692,12 +695,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } @Test - public void testNetSingleUseWithInbound() throws Exception { + public void testNetSingleUseWithInbound() throws Exception { final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -752,12 +755,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } @Test - public void testNioSingleUseWithInbound() throws Exception { + public void testNioSingleUseWithInbound() throws Exception { final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -812,14 +815,13 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } @Test - public void testNioSingleUseWithInboundMany() throws Exception { + public void testNioSingleUseWithInboundMany() throws Exception { final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); final List serverSockets = new ArrayList(); - final ExecutorService exec = Executors.newCachedThreadPool(); - exec.execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100); serverSocket.set(server); @@ -828,7 +830,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final Socket socket = server.accept(); serverSockets.add(socket); final int j = i; - exec.execute(() -> { + this.executor.execute(() -> { semaphore.release(); byte[] b = new byte[9]; try { @@ -843,7 +845,8 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest try { socket.close(); } - catch (IOException e2) { } + catch (IOException e2) { + } } }); } @@ -864,7 +867,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest ccf.setDeserializer(serializer); ccf.setSoTimeout(10000); ccf.setSingleUse(true); - ccf.setTaskExecutor(Executors.newCachedThreadPool()); + ccf.setTaskExecutor(this.executor); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); @@ -902,7 +905,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -973,7 +976,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -1010,7 +1013,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] {newInterceptorFactory()}); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] { newInterceptorFactory() }); ccf.setInterceptorFactoryChain(fc); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -1042,7 +1045,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); serverSocket.set(server); @@ -1099,7 +1102,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + this.executor.execute(() -> { int i = 0; try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index a3f5205bfb..68ff5dcc65 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 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. @@ -68,6 +68,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.TcpOutboundGateway; @@ -782,7 +783,7 @@ public class CachingClientConnectionFactoryTests { invocation.callRealMethod(); String log = invocation.getArgument(0); if (log.startsWith("Response")) { - Executors.newSingleThreadScheduledExecutor() + new SimpleAsyncTaskExecutor() .execute(() -> gate.handleMessage(new GenericMessage<>("bar"))); // hold up the first thread until the second has added its pending reply latch.await(10, TimeUnit.SECONDS); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java index 8b6b342f13..3b0e88554b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 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. @@ -73,7 +73,7 @@ import org.springframework.messaging.support.MessageBuilder; /** * @author Gary Russell - * @author Artyem Bilan + * @author Artem Bilan * * @since 3.0 * @@ -216,6 +216,8 @@ public class ConnectionEventTests { TcpConnectionFailedCorrelationEvent event = (TcpConnectionFailedCorrelationEvent) theEvent.get(); assertEquals("bar", event.getConnectionId()); assertSame(message, ((MessagingException) event.getCause()).getFailedMessage()); + gw.stop(); + scf.stop(); } @Test @@ -260,6 +262,8 @@ public class ConnectionEventTests { messagingException = (MessagingException) event.getCause(); assertSame(message, messagingException.getFailedMessage()); assertEquals("Cannot correlate response - no connection id", messagingException.getMessage()); + gw.stop(); + ccf.stop(); } private void testServerExceptionGuts(AbstractServerConnectionFactory factory) throws Exception { @@ -336,6 +340,8 @@ public class ConnectionEventTests { TcpConnectionFailedEvent event = (TcpConnectionFailedEvent) failEvent.get(); assertSame(e, event.getCause()); } + + ccf.stop(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java index d25174ed48..1ef051a5f4 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.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. @@ -29,6 +29,8 @@ import org.springframework.util.StopWatch; /** * @author Gary Russell + * @author Artem Bilan + * * @since 2.2 * */ @@ -39,7 +41,7 @@ public class ConnectionFactoryShutDownTests { final AbstractConnectionFactory factory = new AbstractConnectionFactory(0) { @Override - public TcpConnection getConnection() throws Exception { + public TcpConnection getConnection() { return null; } @@ -53,7 +55,7 @@ public class ConnectionFactoryShutDownTests { try { while (true) { factory.getTaskExecutor(); - Thread.sleep(100); + Thread.sleep(10); } } catch (MessagingException e1) { @@ -68,7 +70,7 @@ public class ConnectionFactoryShutDownTests { watch.start(); factory.stop(); watch.stop(); - assertTrue("Expected < 10000, was:" + watch.getLastTaskTimeMillis(), watch.getLastTaskTimeMillis() < 10000); + assertTrue("Expected < 10000, was: " + watch.getLastTaskTimeMillis(), watch.getLastTaskTimeMillis() < 10000); assertTrue(latch1.await(10, TimeUnit.SECONDS)); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java index db7ed2df46..a0bd123542 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java @@ -39,7 +39,6 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; @@ -52,6 +51,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean; @@ -233,7 +233,8 @@ public class ConnectionFactoryTests { factory.start(); assertTrue("missing info log", latch1.await(10, TimeUnit.SECONDS)); // stop on a different thread because it waits for the executor - Executors.newSingleThreadExecutor().execute(() -> factory.stop()); + new SimpleAsyncTaskExecutor() + .execute(factory::stop); int n = 0; DirectFieldAccessor accessor = new DirectFieldAccessor(factory); while (n++ < 200 && accessor.getPropertyValue(property) != null) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java index 7858a52720..c252f741cd 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java @@ -35,7 +35,6 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -48,6 +47,7 @@ import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.BridgeHandler; @@ -548,8 +548,9 @@ public class FailoverClientConnectionFactoryTests { } private Holder setupAndStartServers(AbstractServerConnectionFactory server1, - AbstractServerConnectionFactory server2) throws Exception { - Executor exec = Executors.newCachedThreadPool(); + AbstractServerConnectionFactory server2) { + + Executor exec = new SimpleAsyncTaskExecutor(); server1.setTaskExecutor(exec); server2.setTaskExecutor(exec); server1.setBeanName("server1"); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index cbeffd04a2..b920c0c141 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -72,8 +72,11 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.DisposableBean; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; @@ -111,12 +114,14 @@ public class TcpNioConnectionTests { private final ApplicationEventPublisher nullPublisher = mock(ApplicationEventPublisher.class); + private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + @Test public void testWriteTimeout() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch done = new CountDownLatch(1); - final AtomicReference serverSocket = new AtomicReference(); - Executors.newSingleThreadExecutor().execute(() -> { + final AtomicReference serverSocket = new AtomicReference<>(); + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort()); @@ -134,7 +139,7 @@ public class TcpNioConnectionTests { TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", serverSocket.get().getLocalPort()); factory.setApplicationEventPublisher(nullPublisher); - factory.setSoTimeout(1000); + factory.setSoTimeout(100); factory.start(); try { TcpConnection connection = factory.getConnection(); @@ -145,6 +150,7 @@ public class TcpNioConnectionTests { ":" + e.getMessage(), e instanceof SocketTimeoutException); } done.countDown(); + factory.stop(); serverSocket.get().close(); } @@ -152,8 +158,8 @@ public class TcpNioConnectionTests { public void testReadTimeout() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch done = new CountDownLatch(1); - final AtomicReference serverSocket = new AtomicReference(); - Executors.newSingleThreadExecutor().execute(() -> { + final AtomicReference serverSocket = new AtomicReference<>(); + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort()); @@ -173,14 +179,14 @@ public class TcpNioConnectionTests { TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", serverSocket.get().getLocalPort()); factory.setApplicationEventPublisher(nullPublisher); - factory.setSoTimeout(1000); + factory.setSoTimeout(100); factory.start(); try { TcpConnection connection = factory.getConnection(); connection.send(MessageBuilder.withPayload("Test").build()); int n = 0; while (connection.isOpen()) { - Thread.sleep(100); + Thread.sleep(10); if (n++ > 200) { break; } @@ -191,14 +197,15 @@ public class TcpNioConnectionTests { fail("Unexpected exception " + e); } done.countDown(); + factory.stop(); serverSocket.get().close(); } @Test public void testMemoryLeak() throws Exception { final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference serverSocket = new AtomicReference(); - Executors.newSingleThreadExecutor().execute(() -> { + final AtomicReference serverSocket = new AtomicReference<>(); + this.executor.execute(() -> { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort()); @@ -247,7 +254,7 @@ public class TcpNioConnectionTests { TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", 0); factory.setApplicationEventPublisher(nullPublisher); factory.setNioHarvestInterval(100); - Map connections = new HashMap(); + Map connections = new HashMap<>(); SocketChannel chan1 = mock(SocketChannel.class); SocketChannel chan2 = mock(SocketChannel.class); SocketChannel chan3 = mock(SocketChannel.class); @@ -334,6 +341,9 @@ public class TcpNioConnectionTests { catch (ExecutionException e) { assertEquals("Timed out waiting for buffer space", e.getCause().getMessage()); } + finally { + exec.shutdownNow(); + } } @Test @@ -375,6 +385,8 @@ public class TcpNioConnectionTests { }); future.get(60, TimeUnit.SECONDS); assertTrue(messageLatch.await(10, TimeUnit.SECONDS)); + + exec.shutdownNow(); } @Test @@ -463,19 +475,14 @@ public class TcpNioConnectionTests { .getPropertyValue("channelInputStream"); final CountDownLatch latch = new CountDownLatch(1); final byte[] out = new byte[4]; - ExecutorService exec = Executors.newSingleThreadExecutor(); - exec.execute(new Runnable() { - - @Override - public void run() { - try { - stream.read(out); - } - catch (IOException e) { - e.printStackTrace(); - } - latch.countDown(); + this.executor.execute(() -> { + try { + stream.read(out); } + catch (IOException e) { + e.printStackTrace(); + } + latch.countDown(); }); Thread.sleep(1000); assertEquals(0x00, out[0]); @@ -599,11 +606,18 @@ public class TcpNioConnectionTests { assertThat(threadName.get(), containsString("assembler")); factory.stop(); + + cleanupCompositeExecutor(compositeExec); + } + + private void cleanupCompositeExecutor(CompositeExecutor compositeExec) throws Exception { + TestUtils.getPropertyValue(compositeExec, "primaryTaskExecutor", DisposableBean.class).destroy(); + TestUtils.getPropertyValue(compositeExec, "secondaryTaskExecutor", DisposableBean.class).destroy(); } @Test public void testAllMessagesDelivered() throws Exception { - final int numberOfSockets = 100; + final int numberOfSockets = 10; TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(0); factory.setApplicationEventPublisher(nullPublisher); @@ -611,16 +625,11 @@ public class TcpNioConnectionTests { factory.setTaskExecutor(compositeExec); final CountDownLatch latch = new CountDownLatch(numberOfSockets * 4); - factory.registerListener(new TcpListener() { - - @Override - public boolean onMessage(Message message) { - if (!(message instanceof ErrorMessage)) { - latch.countDown(); - } - return false; + factory.registerListener(message -> { + if (!(message instanceof ErrorMessage)) { + latch.countDown(); } - + return false; }); factory.start(); TestingUtilities.waitListening(factory, null); @@ -637,7 +646,7 @@ public class TcpNioConnectionTests { } catch (ConnectException e) { } - Thread.sleep(100); + Thread.sleep(1); } assertTrue("Could not open socket to localhost:" + port, n < 100); sockets[i] = socket; @@ -646,7 +655,7 @@ public class TcpNioConnectionTests { sockets[i].getOutputStream().write("foo1 and...".getBytes()); sockets[i].getOutputStream().flush(); } - Thread.sleep(100); + Thread.sleep(1); for (int i = 0; i < numberOfSockets; i++) { sockets[i].getOutputStream().write(("...foo2\r\nbar1 and...").getBytes()); sockets[i].getOutputStream().flush(); @@ -659,7 +668,7 @@ public class TcpNioConnectionTests { sockets[i].getOutputStream().write("foo3 and...".getBytes()); sockets[i].getOutputStream().flush(); } - Thread.sleep(100); + Thread.sleep(1); for (int i = 0; i < numberOfSockets; i++) { sockets[i].getOutputStream().write(("...foo4\r\nbar3 and...").getBytes()); sockets[i].getOutputStream().flush(); @@ -672,6 +681,8 @@ public class TcpNioConnectionTests { assertTrue("latch is still " + latch.getCount(), latch.await(60, TimeUnit.SECONDS)); factory.stop(); + + cleanupCompositeExecutor(compositeExec); } private CompositeExecutor compositeExecutor() { @@ -791,6 +802,8 @@ public class TcpNioConnectionTests { assertThat(Arrays.asList(stackTrace).toString(), not(containsString("ChannelInputStream.getNextBuffer"))); socket.close(); factory.stop(); + + te.shutdown(); } private void readFully(InputStream is, byte[] buff) throws IOException { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java index 2cf6512ac1..adf7022725 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 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. @@ -38,6 +38,8 @@ import org.springframework.integration.support.MessageBuilder; /** * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class TcpNioConnectionWriteTests { @@ -61,9 +63,10 @@ public class TcpNioConnectionWriteTests { server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(() -> { + AbstractConnectionFactory ccf = null; try { ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); - AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer); + ccf = getClientConnectionFactory(false, port, serializer); TcpConnection connection = ccf.getConnection(); connection.send(MessageBuilder.withPayload(testString.getBytes()).build()); latch.await(10, TimeUnit.SECONDS); @@ -71,6 +74,11 @@ public class TcpNioConnectionWriteTests { catch (Exception e) { e.printStackTrace(); } + finally { + if (ccf != null) { + ccf.stop(); + } + } }); t.setDaemon(true); t.start(); @@ -94,9 +102,10 @@ public class TcpNioConnectionWriteTests { server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(() -> { + AbstractConnectionFactory ccf = null; try { ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); - AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer); + ccf = getClientConnectionFactory(false, port, serializer); TcpConnection connection = ccf.getConnection(); connection.send(MessageBuilder.withPayload(testString.getBytes()).build()); latch.await(10, TimeUnit.SECONDS); @@ -104,6 +113,11 @@ public class TcpNioConnectionWriteTests { catch (Exception e) { e.printStackTrace(); } + finally { + if (ccf != null) { + ccf.stop(); + } + } }); t.setDaemon(true); t.start(); @@ -127,9 +141,10 @@ public class TcpNioConnectionWriteTests { server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(() -> { + AbstractConnectionFactory ccf = null; try { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); - AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer); + ccf = getClientConnectionFactory(false, port, serializer); TcpConnection connection = ccf.getConnection(); connection.send(MessageBuilder.withPayload(testString.getBytes()).build()); latch.await(10, TimeUnit.SECONDS); @@ -137,6 +152,11 @@ public class TcpNioConnectionWriteTests { catch (Exception e) { e.printStackTrace(); } + finally { + if (ccf != null) { + ccf.stop(); + } + } }); t.setDaemon(true); t.start(); @@ -160,9 +180,10 @@ public class TcpNioConnectionWriteTests { server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(() -> { + AbstractConnectionFactory ccf = null; try { ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); - AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer); + ccf = getClientConnectionFactory(true, port, serializer); TcpConnection connection = ccf.getConnection(); connection.send(MessageBuilder.withPayload(testString.getBytes()).build()); latch.await(10, TimeUnit.SECONDS); @@ -170,6 +191,11 @@ public class TcpNioConnectionWriteTests { catch (Exception e) { e.printStackTrace(); } + finally { + if (ccf != null) { + ccf.stop(); + } + } }); t.setDaemon(true); t.start(); @@ -193,9 +219,10 @@ public class TcpNioConnectionWriteTests { server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(() -> { + AbstractConnectionFactory ccf = null; try { ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); - AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer); + ccf = getClientConnectionFactory(true, port, serializer); TcpConnection connection = ccf.getConnection(); connection.send(MessageBuilder.withPayload(testString.getBytes()).build()); latch.await(10, TimeUnit.SECONDS); @@ -203,6 +230,11 @@ public class TcpNioConnectionWriteTests { catch (Exception e) { e.printStackTrace(); } + finally { + if (ccf != null) { + ccf.stop(); + } + } }); t.setDaemon(true); t.start(); @@ -226,9 +258,10 @@ public class TcpNioConnectionWriteTests { server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(() -> { + AbstractConnectionFactory ccf = null; try { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); - AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer); + ccf = getClientConnectionFactory(true, port, serializer); TcpConnection connection = ccf.getConnection(); connection.send(MessageBuilder.withPayload(testString.getBytes()).build()); latch.await(10, TimeUnit.SECONDS); @@ -236,6 +269,11 @@ public class TcpNioConnectionWriteTests { catch (Exception e) { e.printStackTrace(); } + finally { + if (ccf != null) { + ccf.stop(); + } + } }); t.setDaemon(true); t.start(); @@ -251,10 +289,6 @@ public class TcpNioConnectionWriteTests { latch.countDown(); } - /** - * @param is - * @param buff - */ private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { buff[i] = (byte) is.read(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java index 1d38a6116c..293bd9b6de 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java @@ -33,8 +33,7 @@ import java.net.ServerSocket; import java.net.Socket; import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicReference; import javax.net.ServerSocketFactory; @@ -46,6 +45,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.DefaultDeserializer; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.tcp.TcpInboundGateway; import org.springframework.integration.ip.tcp.TcpOutboundGateway; @@ -62,6 +62,7 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell * @author Gavin Gray + * * @since 2.0 */ public class DeserializationTests { @@ -419,7 +420,7 @@ public class DeserializationTests { // eat SocketTimeoutException. Doesn't matter for this test } }; - ExecutorService exec = Executors.newSingleThreadExecutor(); + Executor exec = new SimpleAsyncTaskExecutor(); Message message; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java index ca80f10413..a745d204b5 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java @@ -27,7 +27,6 @@ import java.net.InetSocketAddress; import java.net.MulticastSocket; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -35,6 +34,7 @@ import org.junit.Rule; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -89,7 +89,7 @@ public class DatagramPacketMulticastSendingHandlerTests { e.printStackTrace(); } }; - Executor executor = Executors.newFixedThreadPool(2); + Executor executor = new SimpleAsyncTaskExecutor(); executor.execute(catcher); executor.execute(catcher); assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); @@ -159,7 +159,7 @@ public class DatagramPacketMulticastSendingHandlerTests { e.printStackTrace(); } }; - Executor executor = Executors.newFixedThreadPool(2); + Executor executor = new SimpleAsyncTaskExecutor(); executor.execute(catcher); executor.execute(catcher); assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java index 01fca4e365..92afba7cea 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.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. @@ -24,13 +24,13 @@ import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetSocketAddress; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -39,6 +39,8 @@ import org.springframework.messaging.Message; * @author Mark Fisher * @author Gary Russell * @author Marcin Pilaczynski + * @author Artem Bilan + * * @since 2.0 */ public class DatagramPacketSendingHandlerTests { @@ -50,19 +52,20 @@ public class DatagramPacketSendingHandlerTests { final CountDownLatch received = new CountDownLatch(1); final AtomicInteger testPort = new AtomicInteger(); final CountDownLatch listening = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { - try { - DatagramSocket socket = new DatagramSocket(); - testPort.set(socket.getLocalPort()); - listening.countDown(); - socket.receive(receivedPacket); - received.countDown(); - socket.close(); - } - catch (Exception e) { - e.printStackTrace(); - } - }); + new SimpleAsyncTaskExecutor() + .execute(() -> { + try { + DatagramSocket socket = new DatagramSocket(); + testPort.set(socket.getLocalPort()); + listening.countDown(); + socket.receive(receivedPacket); + received.countDown(); + socket.close(); + } + catch (Exception e) { + e.printStackTrace(); + } + }); assertTrue(listening.await(10, TimeUnit.SECONDS)); UnicastSendingMessageHandler handler = new UnicastSendingMessageHandler("localhost", testPort.get()); @@ -89,31 +92,32 @@ public class DatagramPacketSendingHandlerTests { final CountDownLatch listening = new CountDownLatch(1); final CountDownLatch ackListening = new CountDownLatch(1); final CountDownLatch ackSent = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { - try { - DatagramSocket socket = new DatagramSocket(); - testPort.set(socket.getLocalPort()); - listening.countDown(); - assertTrue(ackListening.await(10, TimeUnit.SECONDS)); - socket.receive(receivedPacket); - socket.close(); - DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); - mapper.setAcknowledge(true); - mapper.setLengthCheck(true); - Message message = mapper.toMessage(receivedPacket); - Object id = message.getHeaders().get(IpHeaders.ACK_ID); - byte[] ack = id.toString().getBytes(); - DatagramPacket ackPack = new DatagramPacket(ack, ack.length, - new InetSocketAddress("localHost", ackPort.get())); - DatagramSocket out = new DatagramSocket(); - out.send(ackPack); - out.close(); - ackSent.countDown(); - } - catch (Exception e) { - e.printStackTrace(); - } - }); + new SimpleAsyncTaskExecutor() + .execute(() -> { + try { + DatagramSocket socket = new DatagramSocket(); + testPort.set(socket.getLocalPort()); + listening.countDown(); + assertTrue(ackListening.await(10, TimeUnit.SECONDS)); + socket.receive(receivedPacket); + socket.close(); + DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); + mapper.setAcknowledge(true); + mapper.setLengthCheck(true); + Message message = mapper.toMessage(receivedPacket); + Object id = message.getHeaders().get(IpHeaders.ACK_ID); + byte[] ack = id.toString().getBytes(); + DatagramPacket ackPack = new DatagramPacket(ack, ack.length, + new InetSocketAddress("localHost", ackPort.get())); + DatagramSocket out = new DatagramSocket(); + out.send(ackPack); + out.close(); + ackSent.countDown(); + } + catch (Exception e) { + e.printStackTrace(); + } + }); listening.await(10000, TimeUnit.MILLISECONDS); UnicastSendingMessageHandler handler = new UnicastSendingMessageHandler("localhost", testPort.get(), true, true, "localhost", 0, 5000); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index 025a9c9708..22ab078fb8 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.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. @@ -31,7 +31,6 @@ import java.net.InetSocketAddress; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -42,6 +41,7 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.ServiceActivatingHandler; @@ -56,6 +56,7 @@ import org.springframework.messaging.SubscribableChannel; * @author Gary Russell * @author Artem Bilan * @author Marcin Pilaczynski + * * @since 2.0 * */ @@ -185,18 +186,19 @@ public class UdpChannelAdapterTests { final CountDownLatch receiverReadyLatch = new CountDownLatch(1); final CountDownLatch replyReceivedLatch = new CountDownLatch(1); //main thread sends the reply using the headers, this thread will receive it - Executors.newSingleThreadExecutor().execute(() -> { - DatagramPacket answer = new DatagramPacket(new byte[2000], 2000); - try { - receiverReadyLatch.countDown(); - socket.receive(answer); - theAnswer.set(answer); - replyReceivedLatch.countDown(); - } - catch (IOException e) { - e.printStackTrace(); - } - }); + new SimpleAsyncTaskExecutor() + .execute(() -> { + DatagramPacket answer = new DatagramPacket(new byte[2000], 2000); + try { + receiverReadyLatch.countDown(); + socket.receive(answer); + theAnswer.set(answer); + replyReceivedLatch.countDown(); + } + catch (IOException e) { + e.printStackTrace(); + } + }); Message receivedMessage = (Message) channel.receive(10000); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); String replyString = "reply:" + System.currentTimeMillis();