From c3b64dc1acc34db5ad6dc2026c2c6e758c742349 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 19 Jan 2018 12:34:57 -0500 Subject: [PATCH] INT-4366: Fix MulticastSendingMessageHandler (#2329) * INT-4366: Fix MulticastSendingMessageHandler JIRA: https://jira.spring.io/browse/INT-4366 Fix race condition in the `MulticastSendingMessageHandler` around `multicastSocket` and super `socket` properties. * Synchronize around `this` and check for the `multicastSocket == null`. This let the `MulticastSendingMessageHandler` to fully configure and prepare the socket for use. * Remove `socket.setInterface(whichNic)` since it is populated by the `InetSocketAddress` ctor before **Cherry-pick to 4.3.x** * Fix thread leaks in TCP/IP tests --- .../udp/MulticastSendingMessageHandler.java | 66 +++++++-------- .../ip/tcp/TcpInboundGatewayTests.java | 55 ++++++------ .../ip/tcp/TcpOutboundGatewayTests.java | 43 +++++----- .../tcp/TcpReceivingChannelAdapterTests.java | 41 +++++---- .../ip/tcp/TcpSendingMessageHandlerTests.java | 68 +++++++-------- .../CachingClientConnectionFactoryTests.java | 5 +- .../connection/ConnectionFactoryTests.java | 5 +- .../FailoverClientConnectionFactoryTests.java | 7 +- .../tcp/connection/TcpNioConnectionTests.java | 81 ++++++++++-------- .../tcp/serializer/DeserializationTests.java | 7 +- ...ramPacketMulticastSendingHandlerTests.java | 6 +- .../DatagramPacketSendingHandlerTests.java | 84 ++++++++++--------- .../ip/udp/UdpChannelAdapterTests.java | 30 +++---- 13 files changed, 262 insertions(+), 236 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index 36c92f137a..d27565ebc8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2016 the original author or authors. + * Copyright 2001-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.messaging.MessageHandler; * determine success. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler { @@ -126,49 +128,45 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler @Override protected DatagramSocket getSocket() throws IOException { - if (this.getTheSocket() == null) { + if (this.multicastSocket == null) { synchronized (this) { - createSocket(); + if (this.multicastSocket == null) { + createSocket(); + } } } - return this.getTheSocket(); + return getTheSocket(); } private void createSocket() throws IOException { - if (this.getTheSocket() == null) { - MulticastSocket socket; - if (this.isAcknowledge()) { - int ackPort = this.getAckPort(); - if (this.localAddress == null) { - socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort); - } - else { - InetAddress whichNic = InetAddress.getByName(this.localAddress); - socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort)); - } - if (getSoReceiveBufferSize() > 0) { - socket.setReceiveBufferSize(this.getSoReceiveBufferSize()); - } - if (logger.isDebugEnabled()) { - logger.debug("Listening for acks on port: " + socket.getLocalPort()); - } - setSocket(socket); - updateAckAddress(); + MulticastSocket socket; + if (isAcknowledge()) { + int ackPort = getAckPort(); + if (this.localAddress == null) { + socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort); } else { - socket = new MulticastSocket(); - setSocket(socket); - } - if (this.timeToLive >= 0) { - socket.setTimeToLive(this.timeToLive); - } - setSocketAttributes(socket); - if (this.localAddress != null) { InetAddress whichNic = InetAddress.getByName(this.localAddress); - socket.setInterface(whichNic); + socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort)); } - this.multicastSocket = socket; + if (getSoReceiveBufferSize() > 0) { + socket.setReceiveBufferSize(getSoReceiveBufferSize()); + } + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + socket.getLocalPort()); + } + setSocket(socket); + updateAckAddress(); } + else { + socket = new MulticastSocket(); + setSocket(socket); + } + if (this.timeToLive >= 0) { + socket.setTimeToLive(this.timeToLive); + } + setSocketAttributes(socket); + this.multicastSocket = socket; } @@ -178,7 +176,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler * @param minAcksForSuccess The minimum number of acks that will represent success. */ public void setMinAcksForSuccess(int minAcksForSuccess) { - this.setAckCounter(minAcksForSuccess); + setAckCounter(minAcksForSuccess); } /** 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..d03afc1b1d 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(); @@ -165,8 +168,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); @@ -220,8 +223,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 +263,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 { @@ -344,7 +347,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 +401,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; })); @@ -442,7 +445,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); @@ -517,12 +520,12 @@ public class TcpOutboundGatewayTests { @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); @@ -667,11 +670,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 +796,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..bed26efc94 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); @@ -215,7 +217,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 +273,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 +326,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 +379,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 +438,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 +497,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 +546,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 +600,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 +647,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 +694,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 +754,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 +814,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 +829,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 +844,8 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest try { socket.close(); } - catch (IOException e2) { } + catch (IOException e2) { + } } }); } @@ -864,7 +866,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 +904,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 +975,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 +1012,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 +1044,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 +1101,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/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..ae58512983 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(); @@ -152,8 +157,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 +178,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; } @@ -197,8 +202,8 @@ public class TcpNioConnectionTests { @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 +252,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 +339,9 @@ public class TcpNioConnectionTests { catch (ExecutionException e) { assertEquals("Timed out waiting for buffer space", e.getCause().getMessage()); } + finally { + exec.shutdownNow(); + } } @Test @@ -375,6 +383,8 @@ public class TcpNioConnectionTests { }); future.get(60, TimeUnit.SECONDS); assertTrue(messageLatch.await(10, TimeUnit.SECONDS)); + + exec.shutdownNow(); } @Test @@ -463,19 +473,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 +604,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 +623,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 +644,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 +653,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 +666,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 +679,8 @@ public class TcpNioConnectionTests { assertTrue("latch is still " + latch.getCount(), latch.await(60, TimeUnit.SECONDS)); factory.stop(); + + cleanupCompositeExecutor(compositeExec); } private CompositeExecutor compositeExecutor() { @@ -791,6 +800,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/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();