From 1dbe351630a735b6b16fd2554f9a9a5580673db9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 15 Jan 2014 12:56:03 -0500 Subject: [PATCH] INT-3212 Backport Fixed CI Build Test https://github.com/spring-projects/spring-integration/commit/405fce672b79a3f6bb8d402c17420188b8b27e2f#diff-cce1a7c963da2228ded1dd0bbea85846 https://build.springsource.org/browse/INT-SI22SPRING30-541/ --- .../ip/tcp/TcpSendingMessageHandlerTests.java | 57 ++++++++++++++++--- 1 file changed, 48 insertions(+), 9 deletions(-) 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 ae853eec5f..e166e50b7d 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 @@ -35,6 +35,7 @@ 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; @@ -48,6 +49,7 @@ import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.serializer.DefaultDeserializer; @@ -97,6 +99,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -148,6 +151,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -209,6 +213,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -262,6 +267,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -312,6 +318,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -365,6 +372,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -418,6 +426,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -474,6 +483,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -523,6 +533,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -576,6 +587,7 @@ public class TcpSendingMessageHandlerTests { final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -620,6 +632,7 @@ public class TcpSendingMessageHandlerTests { final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -664,6 +677,7 @@ public class TcpSendingMessageHandlerTests { final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -721,6 +735,7 @@ public class TcpSendingMessageHandlerTests { final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -778,20 +793,39 @@ public class TcpSendingMessageHandlerTests { final Semaphore semaphore = new Semaphore(0); final AtomicBoolean done = new AtomicBoolean(); final List serverSockets = new ArrayList(); - Executors.newSingleThreadExecutor().execute(new Runnable() { + final ExecutorService exec = Executors.newCachedThreadPool(); + exec.execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); latch.countDown(); for (int i = 0; i < 100; i++) { - Socket socket = server.accept(); + final Socket socket = server.accept(); serverSockets.add(socket); - semaphore.release(); - byte[] b = new byte[9]; - readFully(socket.getInputStream(), b); - b = ("Reply" + i + "\r\n").getBytes(); - socket.getOutputStream().write(b); - socket.close(); + final int j = i; + exec.execute(new Runnable() { + + @Override + public void run() { + semaphore.release(); + byte[] b = new byte[9]; + try { + readFully(socket.getInputStream(), b); + b = ("Reply" + j + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + catch (IOException e) { + e.printStackTrace(); + } + finally { + try { + socket.close(); + } + catch (IOException e) { } + } + } + }); } server.close(); } catch (Exception e) { @@ -807,7 +841,7 @@ public class TcpSendingMessageHandlerTests { ccf.setDeserializer(serializer); ccf.setSoTimeout(10000); ccf.setSingleUse(true); - ccf.setTaskExecutor(Executors.newFixedThreadPool(100)); + ccf.setTaskExecutor(Executors.newCachedThreadPool()); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); @@ -845,6 +879,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -913,6 +948,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -979,6 +1015,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -1037,6 +1074,7 @@ public class TcpSendingMessageHandlerTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { int i = 0; try { @@ -1109,6 +1147,7 @@ public class TcpSendingMessageHandlerTests { AbstractConnectionFactory mockCcf = mock(AbstractClientConnectionFactory.class); Mockito.doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { throw new SocketException("Failed to connect"); }