From fb766a951cb097ad1a7847a958630428f098a6b6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 29 Oct 2015 15:01:28 -0400 Subject: [PATCH] TCP: Tests: Let OS Choose Test Port https://build.spring.io/browse/INT-MJATS41-434 BindException. Avoid using `findAvailableServerSocket()` where easily possible. Some still remain, but they would require major rework of the tests. Also fix a race in `TcpOutboundGatewayTests.testGoodNetGWTimeoutGuts()`. Although an `AtomicBoolean` was used, both threads might still see the lower (500ms) timeout. Polishing --- .../ip/tcp/FactoryStopStartTests.java | 7 +- .../ip/tcp/TcpInboundGatewayTests.java | 51 +-- .../ip/tcp/TcpOutboundGatewayTests.java | 134 +++++--- .../tcp/TcpReceivingChannelAdapterTests.java | 94 +++--- .../ip/tcp/TcpSendingMessageHandlerTests.java | 302 +++++++++++------- .../FailoverClientConnectionFactoryTests.java | 16 +- .../ip/tcp/connection/SocketSupportTests.java | 101 ++++-- .../connection/TcpNioConnectionReadTests.java | 92 ++++-- .../tcp/connection/TcpNioConnectionTests.java | 57 ++-- .../TcpNioConnectionWriteTests.java | 33 +- .../tcp/serializer/DeserializationTests.java | 45 ++- .../ip/tcp/serializer/SerializationTests.java | 23 +- 12 files changed, 561 insertions(+), 394 deletions(-) diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java index 8bfb9b63cd..11971592a5 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,9 +16,9 @@ package org.springframework.integration.ip.tcp; import org.junit.Test; + import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; -import org.springframework.integration.test.util.SocketUtils; /** @@ -30,8 +30,7 @@ public class FactoryStopStartTests { @Test public void testRestart() { - int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory factory = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory factory = new TcpNetServerConnectionFactory(0); factory.setSoTimeout(10000); factory.start(); factory.stop(); 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 e5f3c1ad7a..d485dc6f92 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-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,6 +31,7 @@ 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; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; @@ -47,7 +48,6 @@ import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionF import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.util.TestingUtilities; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -61,14 +61,14 @@ public class TcpInboundGatewayTests { @Test public void testNetSingle() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); scf.setSingleUse(true); TcpInboundGateway gateway = new TcpInboundGateway(); gateway.setConnectionFactory(scf); gateway.setBeanFactory(mock(BeanFactory.class)); scf.start(); TestingUtilities.waitListening(scf, 20000L); + int port = scf.getPort(); final QueueChannel channel = new QueueChannel(); gateway.setRequestChannel(channel); ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); @@ -93,13 +93,13 @@ public class TcpInboundGatewayTests { @Test public void testNetNotSingle() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); scf.setSingleUse(false); TcpInboundGateway gateway = new TcpInboundGateway(); gateway.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, 20000L); + int port = scf.getPort(); final QueueChannel channel = new QueueChannel(); gateway.setRequestChannel(channel); gateway.setBeanFactory(mock(BeanFactory.class)); @@ -118,18 +118,7 @@ public class TcpInboundGatewayTests { @Test public void testNetClientMode() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); - ccf.setSingleUse(false); - TcpInboundGateway gateway = new TcpInboundGateway(); - gateway.setConnectionFactory(ccf); - final QueueChannel channel = new QueueChannel(); - gateway.setRequestChannel(channel); - gateway.setClientMode(true); - gateway.setRetryInterval(10000); - gateway.setBeanFactory(mock(BeanFactory.class)); - gateway.afterPropertiesSet(); - ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); + final AtomicInteger port = new AtomicInteger(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); @@ -138,7 +127,8 @@ public class TcpInboundGatewayTests { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10); + 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()); @@ -161,6 +151,17 @@ public class TcpInboundGatewayTests { } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port.get()); + ccf.setSingleUse(false); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(ccf); + final QueueChannel channel = new QueueChannel(); + gateway.setRequestChannel(channel); + gateway.setClientMode(true); + gateway.setRetryInterval(10000); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.afterPropertiesSet(); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.setPoolSize(1); taskScheduler.initialize(); @@ -180,13 +181,13 @@ public class TcpInboundGatewayTests { @Test public void testNioSingle() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); scf.setSingleUse(true); TcpInboundGateway gateway = new TcpInboundGateway(); gateway.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, 20000L); + int port = scf.getPort(); final QueueChannel channel = new QueueChannel(); gateway.setRequestChannel(channel); gateway.setBeanFactory(mock(BeanFactory.class)); @@ -212,13 +213,13 @@ public class TcpInboundGatewayTests { @Test public void testNioNotSingle() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); scf.setSingleUse(false); TcpInboundGateway gateway = new TcpInboundGateway(); gateway.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, 20000L); + int port = scf.getPort(); final QueueChannel channel = new QueueChannel(); gateway.setRequestChannel(channel); gateway.setBeanFactory(mock(BeanFactory.class)); @@ -240,8 +241,7 @@ public class TcpInboundGatewayTests { @Test public void testErrorFlow() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); scf.setSingleUse(true); TcpInboundGateway gateway = new TcpInboundGateway(); gateway.setConnectionFactory(scf); @@ -257,6 +257,7 @@ public class TcpInboundGatewayTests { gateway.setErrorChannel(errorChannel); scf.start(); TestingUtilities.waitListening(scf, 20000L); + int port = scf.getPort(); final SubscribableChannel channel = new DirectChannel(); gateway.setRequestChannel(channel); gateway.setBeanFactory(mock(BeanFactory.class)); 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 364c0189bc..eff34615a9 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 @@ -78,7 +78,6 @@ import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionF import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.rule.Log4jLevelAdjuster; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -99,8 +98,6 @@ public class TcpOutboundGatewayTests { @Test public void testGoodNetSingle() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final AtomicReference serverSocket = new AtomicReference(); @@ -108,7 +105,7 @@ public class TcpOutboundGatewayTests { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100); serverSocket.set(server); latch.countDown(); List sockets = new ArrayList(); @@ -129,12 +126,14 @@ public class TcpOutboundGatewayTests { } } }); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(true); ccf.start(); - assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); TcpOutboundGateway gateway = new TcpOutboundGateway(); gateway.setConnectionFactory(ccf); QueueChannel replyChannel = new QueueChannel(); @@ -166,14 +165,15 @@ public class TcpOutboundGatewayTests { @Test public void testGoodNetMultiplex() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference serverSocket = new AtomicReference(); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10); + serverSocket.set(server); latch.countDown(); int i = 0; Socket socket = server.accept(); @@ -190,13 +190,14 @@ public class TcpOutboundGatewayTests { } } }); - AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(false); ccf.start(); - assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); TcpOutboundGateway gateway = new TcpOutboundGateway(); gateway.setConnectionFactory(ccf); QueueChannel replyChannel = new QueueChannel(); @@ -216,18 +217,20 @@ public class TcpOutboundGatewayTests { } done.set(true); gateway.stop(); + serverSocket.get().close(); } @Test public void testGoodNetTimeout() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference serverSocket = new AtomicReference(); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); int i = 0; Socket socket = server.accept(); @@ -238,20 +241,22 @@ public class TcpOutboundGatewayTests { Thread.sleep(1000); oos.writeObject("Reply" + (i++)); } - } catch (Exception e) { + } + catch (Exception e) { if (!done.get()) { e.printStackTrace(); } } } }); - AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(false); ccf.start(); - assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); final TcpOutboundGateway gateway = new TcpOutboundGateway(); gateway.setConnectionFactory(ccf); gateway.setRequestTimeout(1); @@ -297,23 +302,28 @@ public class TcpOutboundGatewayTests { } done.set(true); gateway.stop(); + serverSocket.get().close(); } @Test public void testGoodNetGWTimeout() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = buildCF(port); ccf.start(); - testGoodNetGWTimeoutGuts(port, ccf); + testGoodNetGWTimeoutGuts(port, ccf, serverSocket); + serverSocket.close(); } @Test public void testGoodNetGWTimeoutCached() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = buildCF(port); CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(ccf, 1); cccf.start(); - testGoodNetGWTimeoutGuts(port, cccf); + testGoodNetGWTimeoutGuts(port, cccf, serverSocket); + serverSocket.close(); } private AbstractClientConnectionFactory buildCF(final int port) { @@ -331,8 +341,8 @@ public class TcpOutboundGatewayTests { * own response, not that for the first. * @throws Exception */ - private void testGoodNetGWTimeoutGuts(final int port, AbstractClientConnectionFactory ccf) - throws InterruptedException { + private void testGoodNetGWTimeoutGuts(final int port, AbstractClientConnectionFactory ccf, + final ServerSocket server) throws InterruptedException { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); /* @@ -347,7 +357,6 @@ public class TcpOutboundGatewayTests { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); int i = 0; while (!done.get()) { @@ -397,11 +406,13 @@ public class TcpOutboundGatewayTests { @Override public Object answer(InvocationOnMock invocation) throws Throwable { - if (!remoteTimeoutUsed.getAndSet(true)) { + if (remoteTimeoutUsed.getAndSet(true)) { // increase the timeout after the first send - gateway.setRemoteTimeout(10000); + return 10000L; + } + else { + return 500L; } - return invocation.callRealMethod(); } }).when(remoteTimeoutExpression) @@ -458,7 +469,7 @@ public class TcpOutboundGatewayTests { @Test public void testCachingFailover() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final CountDownLatch serverLatch = new CountDownLatch(1); @@ -468,7 +479,8 @@ public class TcpOutboundGatewayTests { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); while (!done.get()) { Socket socket = server.accept(); @@ -504,7 +516,8 @@ public class TcpOutboundGatewayTests { when(factory1.getConnection()).thenReturn(mockConn1); doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); - AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port); + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); factory2.setSerializer(new DefaultSerializer()); factory2.setDeserializer(new DefaultDeserializer()); factory2.setSoTimeout(10000); @@ -535,11 +548,12 @@ public class TcpOutboundGatewayTests { done.set(true); gateway.stop(); verify(mockConn1).send(Mockito.any(Message.class)); + serverSocket.get().close(); } @Test public void testFailoverCached() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final CountDownLatch serverLatch = new CountDownLatch(1); @@ -549,7 +563,8 @@ public class TcpOutboundGatewayTests { @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); while (!done.get()) { Socket socket = server.accept(); @@ -587,7 +602,8 @@ public class TcpOutboundGatewayTests { doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 1); - AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port); + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); factory2.setSerializer(new DefaultSerializer()); factory2.setDeserializer(new DefaultDeserializer()); factory2.setSoTimeout(10000); @@ -619,6 +635,7 @@ public class TcpOutboundGatewayTests { done.set(true); gateway.stop(); verify(mockConn1).send(Mockito.any(Message.class)); + serverSocket.get().close(); } public TcpConnectionSupport makeMockConnection() { @@ -629,31 +646,36 @@ public class TcpOutboundGatewayTests { @Test public void testNetGWPropagatesSocketClose() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(false); ccf.start(); - testGWPropagatesSocketCloseGuts(port, ccf); + testGWPropagatesSocketCloseGuts(port, ccf, serverSocket); + serverSocket.close(); } @Test public void testNioGWPropagatesSocketClose() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(false); ccf.start(); - testGWPropagatesSocketCloseGuts(port, ccf); + testGWPropagatesSocketCloseGuts(port, ccf, serverSocket); + serverSocket.close(); } @Test public void testCachedGWPropagatesSocketClose() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -661,12 +683,14 @@ public class TcpOutboundGatewayTests { ccf.setSingleUse(false); CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(ccf, 1); cccf.start(); - testGWPropagatesSocketCloseGuts(port, cccf); + testGWPropagatesSocketCloseGuts(port, cccf, serverSocket); + serverSocket.close(); } @Test public void testFailoverGWPropagatesSocketClose() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -675,10 +699,12 @@ public class TcpOutboundGatewayTests { FailoverClientConnectionFactory focf = new FailoverClientConnectionFactory( Collections.singletonList(ccf)); focf.start(); - testGWPropagatesSocketCloseGuts(port, focf); + testGWPropagatesSocketCloseGuts(port, focf, serverSocket); + serverSocket.close(); } - private void testGWPropagatesSocketCloseGuts(final int port, AbstractClientConnectionFactory ccf) throws Exception { + private void testGWPropagatesSocketCloseGuts(final int port, AbstractClientConnectionFactory ccf, + final ServerSocket server) throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final AtomicReference lastReceived = new AtomicReference(); @@ -690,7 +716,6 @@ public class TcpOutboundGatewayTests { public void run() { List sockets = new ArrayList(); try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); while (!done.get()) { Socket socket = server.accept(); @@ -752,54 +777,62 @@ public class TcpOutboundGatewayTests { @Test public void testNetGWPropagatesSocketTimeout() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(100); ccf.setSingleUse(false); ccf.start(); - testGWPropagatesSocketTimeoutGuts(port, ccf); + testGWPropagatesSocketTimeoutGuts(port, ccf, serverSocket); + serverSocket.close(); } @Test public void testNioGWPropagatesSocketTimeout() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(100); ccf.setSingleUse(false); ccf.start(); - testGWPropagatesSocketTimeoutGuts(port, ccf); + testGWPropagatesSocketTimeoutGuts(port, ccf, serverSocket); + serverSocket.close(); } @Test public void testNetGWPropagatesSocketTimeoutSingleUse() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(100); ccf.setSingleUse(true); ccf.start(); - testGWPropagatesSocketTimeoutGuts(port, ccf); + testGWPropagatesSocketTimeoutGuts(port, ccf, serverSocket); + serverSocket.close(); } @Test public void testNioGWPropagatesSocketTimeoutSingleUse() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(100); ccf.setSingleUse(true); ccf.start(); - testGWPropagatesSocketTimeoutGuts(port, ccf); + testGWPropagatesSocketTimeoutGuts(port, ccf, serverSocket); + serverSocket.close(); } - private void testGWPropagatesSocketTimeoutGuts(final int port, AbstractClientConnectionFactory ccf) - throws Exception { + private void testGWPropagatesSocketTimeoutGuts(final int port, AbstractClientConnectionFactory ccf, + final ServerSocket server) throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -809,7 +842,6 @@ public class TcpOutboundGatewayTests { public void run() { List sockets = new ArrayList(); try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); while (!done.get()) { sockets.add(server.accept()); 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 4731b66291..758892ea04 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-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -36,6 +36,7 @@ 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; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; @@ -57,7 +58,6 @@ import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionF import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.TestingUtilities; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.messaging.Message; import org.springframework.messaging.SubscribableChannel; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -69,8 +69,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNet() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -79,6 +78,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); adapter.setBeanFactory(mock(BeanFactory.class)); @@ -97,7 +97,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNetClientMode() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -105,7 +105,8 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10); + 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()); @@ -120,7 +121,9 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe } } }); - AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -133,7 +136,6 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setOutputChannel(channel); adapter.setBeanFactory(mock(BeanFactory.class)); adapter.afterPropertiesSet(); - assertTrue(latch1.await(10, TimeUnit.SECONDS)); adapter.setRetryInterval(10000); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.setPoolSize(1); @@ -151,12 +153,12 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.stop(); latch2.countDown(); ccf.stop(); + serverSocket.get().close(); } @Test public void testNio() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -166,6 +168,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); @@ -186,8 +189,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNetShared() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -200,6 +202,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); socket.setSoTimeout(2000); socket.getOutputStream().write("Test\r\n".getBytes()); @@ -220,8 +223,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNioShared() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -234,6 +236,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); socket.setSoTimeout(2000); socket.getOutputStream().write("Test\r\n".getBytes()); @@ -254,8 +257,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNetSingleNoOutbound() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -265,6 +267,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); @@ -286,8 +289,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNioSingleNoOutbound() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -297,6 +299,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); @@ -328,8 +331,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNetSingleShared() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -343,6 +345,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); socket1.setSoTimeout(2000); socket1.getOutputStream().write("Test1\r\n".getBytes()); @@ -365,8 +368,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNioSingleShared() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -380,6 +382,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); socket1.setSoTimeout(2000); socket1.getOutputStream().write("Test1\r\n".getBytes()); @@ -402,8 +405,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNioSingleSharedMany() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -420,6 +422,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); List sockets = new LinkedList(); for (int i = 100; i < 200; i++) { Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); @@ -442,59 +445,53 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testNetInterceptors() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); - interceptorsGuts(port, scf); + interceptorsGuts(scf); scf.stop(); } @Test public void testNetSingleNoOutboundInterceptors() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); - singleNoOutboundInterceptorsGuts(port, scf); + singleNoOutboundInterceptorsGuts(scf); scf.stop(); } @Test public void testNetSingleSharedInterceptors() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); - singleSharedInterceptorsGuts(port, scf); + singleSharedInterceptorsGuts(scf); scf.stop(); } @Test public void testNioInterceptors() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); - interceptorsGuts(port, scf); + interceptorsGuts(scf); scf.stop(); } @Test public void testNioSingleNoOutboundInterceptors() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); - singleNoOutboundInterceptorsGuts(port, scf); + singleNoOutboundInterceptorsGuts(scf); scf.stop(); } @Test public void testNioSingleSharedInterceptors() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); noopPublisher(scf); - singleSharedInterceptorsGuts(port, scf); + singleSharedInterceptorsGuts(scf); scf.stop(); } - private void interceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { + private void interceptorsGuts(AbstractServerConnectionFactory scf) throws Exception { scf.setSerializer(new DefaultSerializer()); scf.setDeserializer(new DefaultDeserializer()); scf.setSingleUse(false); @@ -509,6 +506,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe scf.setSoTimeout(10000); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); @@ -530,7 +528,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe assertTrue(results.contains("Test2")); } - private void singleNoOutboundInterceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { + private void singleNoOutboundInterceptorsGuts(AbstractServerConnectionFactory scf) throws Exception { scf.setSerializer(new DefaultSerializer()); scf.setDeserializer(new DefaultDeserializer()); scf.setSingleUse(true); @@ -545,6 +543,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); @@ -573,7 +572,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe assertTrue(results.contains("Test2")); } - private void singleSharedInterceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { + private void singleSharedInterceptorsGuts(AbstractServerConnectionFactory scf) throws Exception { scf.setSerializer(new DefaultSerializer()); scf.setDeserializer(new DefaultDeserializer()); scf.setSingleUse(true); @@ -592,6 +591,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); socket1.setSoTimeout(60000); new ObjectOutputStream(socket1.getOutputStream()).writeObject("Hello"); @@ -621,8 +621,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe @Test public void testException() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(0); noopPublisher(scf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); @@ -631,6 +630,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe adapter.setConnectionFactory(scf); scf.start(); TestingUtilities.waitListening(scf, null); + int port = scf.getPort(); SubscribableChannel channel = new DirectChannel(); adapter.setOutputChannel(channel); ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService()); 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 6354ed9334..3b0accc2c9 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 @@ -40,6 +40,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.net.ServerSocketFactory; @@ -68,7 +69,6 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSe import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; @@ -95,14 +95,16 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest @Test public void testNetCrLf() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -119,8 +121,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -133,7 +138,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -144,18 +148,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertEquals("Reply2", new String((byte[]) mOut.getPayload())); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetCrLfClientMode() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -172,8 +179,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -185,7 +195,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.setClientMode(true); handler.setRetryInterval(10000); handler.setBeanFactory(mock(BeanFactory.class)); @@ -209,18 +218,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest handler.start(); handler.stop(); adapter.stop(); + serverSocket.get().close(); } @Test public void testNioCrLf() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -236,8 +248,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -250,7 +265,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Set results = new HashSet(); @@ -264,18 +278,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertTrue(results.remove("Reply2")); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetStxEtx() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -291,8 +308,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); ccf.setSerializer(serializer); @@ -305,7 +325,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -316,18 +335,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertEquals("Reply2", new String((byte[]) mOut.getPayload())); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioStxEtx() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -343,8 +365,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); ccf.setSerializer(serializer); @@ -357,7 +382,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Set results = new HashSet(); @@ -371,18 +395,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertTrue(results.remove("Reply2")); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetLength() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -401,8 +428,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); ccf.setSerializer(serializer); @@ -415,7 +445,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -426,18 +455,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertEquals("Reply2", new String((byte[]) mOut.getPayload())); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioLength() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -450,14 +482,18 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest b = ("\u0000\u0000\u0000\u0006Reply" + (++i)).getBytes(); socket.getOutputStream().write(b); } - } catch (Exception e) { + } + catch (Exception e) { if (!done.get()) { e.printStackTrace(); } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); ccf.setSerializer(serializer); @@ -470,7 +506,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Set results = new HashSet(); @@ -484,18 +519,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertTrue(results.remove("Reply2")); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetSerial() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -511,8 +549,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -524,7 +565,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -535,18 +575,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertEquals("Reply2", mOut.getPayload()); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioSerial() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -562,8 +605,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -575,7 +621,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Set results = new HashSet(); @@ -589,19 +634,22 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertTrue(results.remove("Reply2")); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetSingleUseNoInbound() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + 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(new Runnable() { + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); for (int i = 0; i < 2; i++) { Socket socket = server.accept(); @@ -618,8 +666,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -629,25 +680,27 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest ccf.setSingleUse(true); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS)); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioSingleUseNoInbound() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + 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(new Runnable() { + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); for (int i = 0; i < 2; i++) { Socket socket = server.accept(); @@ -664,8 +717,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -675,25 +731,27 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest ccf.setSingleUse(true); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test.1").build()); handler.handleMessage(MessageBuilder.withPayload("Test.2").build()); assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS)); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetSingleUseWithInbound() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + 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(new Runnable() { + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); for (int i = 1; i < 3; i++) { Socket socket = server.accept(); @@ -711,8 +769,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -726,7 +787,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); @@ -740,19 +800,22 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertTrue(replies.remove("Reply2")); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioSingleUseWithInbound() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + 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(new Runnable() { + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); for (int i = 1; i < 3; i++) { Socket socket = server.accept(); @@ -770,8 +833,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -785,7 +851,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); @@ -799,21 +864,24 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertTrue(replies.remove("Reply2")); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioSingleUseWithInboundMany() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + 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(new Runnable() { + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100); + serverSocket.set(server); latch.countDown(); for (int i = 0; i < 100; i++) { final Socket socket = server.accept(); @@ -843,14 +911,18 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest }); } server.close(); - } catch (Exception e) { + } + catch (Exception e) { if (!done.get()) { e.printStackTrace(); } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); @@ -865,7 +937,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); int i = 0; try { for (i = 100; i < 200; i++) { @@ -887,18 +958,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetNegotiate() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override + + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 0; @@ -928,8 +1002,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -947,7 +1024,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -958,18 +1034,21 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertEquals("Reply2", mOut.getPayload()); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioNegotiate() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); 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); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); int i = 100; @@ -988,14 +1067,18 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest in = ois.readObject(); oos.writeObject("Reply" + (i++)); } - } catch (Exception e) { + } + catch (Exception e) { if (!done.get()) { e.printStackTrace(); } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -1010,7 +1093,6 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); for (int i = 0; i < 1000; i++) { handler.handleMessage(MessageBuilder.withPayload("Test").build()); } @@ -1026,38 +1108,37 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNetNegotiateSingleNoListen() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { - @Override + + @Override public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); - int i = 0; ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); - Object in = null; ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); - if (i == 0) { - in = ois.readObject(); - logger.debug("read object: " + in); - oos.writeObject("world!"); - ois = new ObjectInputStream(socket.getInputStream()); - oos = new ObjectOutputStream(socket.getOutputStream()); - in = ois.readObject(); - logger.debug("read object: " + in); - oos.writeObject("world!"); - ois = new ObjectInputStream(socket.getInputStream()); - oos = new ObjectOutputStream(socket.getOutputStream()); - } + Object in = ois.readObject(); + logger.debug("read object: " + in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); in = ois.readObject(); - oos.writeObject("Reply" + (++i)); + logger.debug("read object: " + in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + in = ois.readObject(); + oos.writeObject("Reply"); socket.close(); server.close(); } @@ -1067,8 +1148,11 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest } } } + }); - AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -1083,52 +1167,54 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test public void testNioNegotiateSingleNoListen() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); + final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { int i = 0; try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); - Object in = null; ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); - if (i == 0) { - in = ois.readObject(); - logger.debug("read object: " + in); - oos.writeObject("world!"); - ois = new ObjectInputStream(socket.getInputStream()); - oos = new ObjectOutputStream(socket.getOutputStream()); - in = ois.readObject(); - logger.debug("read object: " + in); - oos.writeObject("world!"); - ois = new ObjectInputStream(socket.getInputStream()); - oos = new ObjectOutputStream(socket.getOutputStream()); - } + Object in = ois.readObject(); + logger.debug("read object: " + in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); in = ois.readObject(); + logger.debug("read object: " + in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); oos.writeObject("Reply" + (++i)); socket.close(); server.close(); - } catch (Exception e) { + } + catch (Exception e) { if (i == 0) { e.printStackTrace(); } } } + }); - AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); noopPublisher(ccf); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); @@ -1143,10 +1229,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); - assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); done.set(true); ccf.stop(); + serverSocket.get().close(); } @Test @@ -1175,6 +1261,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest public Object answer(InvocationOnMock invocation) throws Throwable { throw new SocketException("Failed to connect"); } + }).when(mockCcf).getConnection(); handler.setConnectionFactory(mockCcf); try { @@ -1188,4 +1275,5 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest assertEquals("Failed to connect", e.getCause().getMessage()); } } + } 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 cdf740571d..682f0da848 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 @@ -311,8 +311,7 @@ public class FailoverClientConnectionFactoryTests { @Test public void testFailoverCachedRealClose() throws Exception { - int port1 = SocketUtils.findAvailableServerSocket(); - TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1); + TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(0); server1.setBeanName("server1"); final CountDownLatch latch1 = new CountDownLatch(3); server1.registerListener(new TcpListener() { @@ -325,8 +324,8 @@ public class FailoverClientConnectionFactoryTests { }); server1.start(); TestingUtilities.waitListening(server1, 10000L); - int port2 = SocketUtils.findAvailableServerSocket(); - TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2); + int port1 = server1.getPort(); + TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(0); server2.setBeanName("server2"); final CountDownLatch latch2 = new CountDownLatch(2); server2.registerListener(new TcpListener() { @@ -339,6 +338,7 @@ public class FailoverClientConnectionFactoryTests { }); server2.start(); TestingUtilities.waitListening(server2, 10000L); + int port2 = server2.getPort(); AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("localhost", port1); factory1.setBeanName("client1"); factory1.registerListener(new TcpListener() { @@ -407,8 +407,7 @@ public class FailoverClientConnectionFactoryTests { @Test public void testFailoverCachedRealBadHost() throws Exception { - int port1 = SocketUtils.findAvailableServerSocket(); - TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1); + TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(0); server1.setBeanName("server1"); final CountDownLatch latch1 = new CountDownLatch(3); server1.registerListener(new TcpListener() { @@ -421,8 +420,8 @@ public class FailoverClientConnectionFactoryTests { }); server1.start(); TestingUtilities.waitListening(server1, 10000L); - int port2 = SocketUtils.findAvailableServerSocket(); - TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2); + int port1 = server1.getPort(); + TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(0); server2.setBeanName("server2"); final CountDownLatch latch2 = new CountDownLatch(2); server2.registerListener(new TcpListener() { @@ -435,6 +434,7 @@ public class FailoverClientConnectionFactoryTests { }); server2.start(); TestingUtilities.waitListening(server2, 10000L); + int port2 = server2.getPort(); AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("junkjunk", port1); factory1.setBeanName("client1"); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java index cdf4c23ac6..69158da309 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java @@ -47,7 +47,6 @@ import org.mockito.stubbing.Answer; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.TestingUtilities; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; @@ -55,7 +54,6 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell * @since 2.2 - * */ public class SocketSupportTests { @@ -99,13 +97,16 @@ public class SocketSupportTests { when(factory.createServerSocket(0, 5)).thenReturn(serverSocket); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - when(serverSocket.accept()).thenReturn(socket).then(new Answer (){ + when(serverSocket.accept()).thenReturn(socket).then(new Answer() { + @Override public Socket answer(InvocationOnMock invocation) throws Throwable { latch1.countDown(); latch2.await(10, TimeUnit.SECONDS); return null; - }}); + } + + }); TcpSocketSupport socketSupport = mock(TcpSocketSupport.class); TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(0); @@ -123,46 +124,55 @@ public class SocketSupportTests { @Test public void testNioClientAndServer() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - TcpNioClientConnectionFactory clientConnectionFactory = new TcpNioClientConnectionFactory("localhost", port); - final AtomicInteger ppSocketCountClient = new AtomicInteger(); - final AtomicInteger ppServerSocketCountClient = new AtomicInteger(); - TcpSocketSupport clientSocketSupport = new TcpSocketSupport() { - @Override - public void postProcessSocket(Socket socket) { - ppSocketCountClient.incrementAndGet(); - } - @Override - public void postProcessServerSocket(ServerSocket serverSocket) { - ppServerSocketCountClient.incrementAndGet(); - } - }; - clientConnectionFactory.setTcpSocketSupport(clientSocketSupport); - clientConnectionFactory.start(); - TcpNioServerConnectionFactory serverConnectionFactory = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory serverConnectionFactory = new TcpNioServerConnectionFactory(0); serverConnectionFactory.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { return false; } + }); final AtomicInteger ppSocketCountServer = new AtomicInteger(); final AtomicInteger ppServerSocketCountServer = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(1); TcpSocketSupport serverSocketSupport = new TcpSocketSupport() { + @Override public void postProcessSocket(Socket socket) { ppSocketCountServer.incrementAndGet(); latch.countDown(); } + @Override public void postProcessServerSocket(ServerSocket serverSocket) { ppServerSocketCountServer.incrementAndGet(); } + }; serverConnectionFactory.setTcpSocketSupport(serverSocketSupport); serverConnectionFactory.start(); TestingUtilities.waitListening(serverConnectionFactory, null); + + TcpNioClientConnectionFactory clientConnectionFactory = new TcpNioClientConnectionFactory("localhost", + serverConnectionFactory.getPort()); + final AtomicInteger ppSocketCountClient = new AtomicInteger(); + final AtomicInteger ppServerSocketCountClient = new AtomicInteger(); + TcpSocketSupport clientSocketSupport = new TcpSocketSupport() { + + @Override + public void postProcessSocket(Socket socket) { + ppSocketCountClient.incrementAndGet(); + } + + @Override + public void postProcessServerSocket(ServerSocket serverSocket) { + ppServerSocketCountClient.incrementAndGet(); + } + + }; + clientConnectionFactory.setTcpSocketSupport(clientSocketSupport); + clientConnectionFactory.start(); clientConnectionFactory.getConnection().send(new GenericMessage("Hello, world!")); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertEquals(0, ppServerSocketCountClient.get()); @@ -170,6 +180,9 @@ public class SocketSupportTests { assertEquals(1, ppServerSocketCountServer.get()); assertEquals(1, ppSocketCountServer.get()); + + clientConnectionFactory.stop(); + serverConnectionFactory.stop(); } /* @@ -273,18 +286,21 @@ Certificate fingerprints: TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0); TcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks", "test.truststore.ks", "secret", "secret"); - DefaultTcpNetSSLSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(sslContextSupport); + DefaultTcpNetSSLSocketFactorySupport tcpSocketFactorySupport = + new DefaultTcpNetSSLSocketFactorySupport(sslContextSupport); tcpSocketFactorySupport.afterPropertiesSet(); server.setTcpSocketFactorySupport(tcpSocketFactorySupport); final List> messages = new ArrayList>(); final CountDownLatch latch = new CountDownLatch(1); server.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { messages.add(message); latch.countDown(); return false; } + }); server.setMapper(new SSLMapper()); server.start(); @@ -308,26 +324,30 @@ Certificate fingerprints: TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0); TcpSSLContextSupport serverSslContextSupport = new DefaultTcpSSLContextSupport("server.ks", "server.truststore.ks", "secret", "secret"); - DefaultTcpNetSSLSocketFactorySupport serverTcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(serverSslContextSupport); + DefaultTcpNetSSLSocketFactorySupport serverTcpSocketFactorySupport = + new DefaultTcpNetSSLSocketFactorySupport(serverSslContextSupport); serverTcpSocketFactorySupport.afterPropertiesSet(); server.setTcpSocketFactorySupport(serverTcpSocketFactorySupport); final List> messages = new ArrayList>(); final CountDownLatch latch = new CountDownLatch(1); server.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { messages.add(message); latch.countDown(); return false; } + }); server.start(); TestingUtilities.waitListening(server, null); TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", server.getPort()); - TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks", "client.truststore.ks", - "secret", "secret"); - DefaultTcpNetSSLSocketFactorySupport clientTcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(clientSslContextSupport); + TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks", + "client.truststore.ks", "secret", "secret"); + DefaultTcpNetSSLSocketFactorySupport clientTcpSocketFactorySupport = + new DefaultTcpNetSSLSocketFactorySupport(clientSslContextSupport); clientTcpSocketFactorySupport.afterPropertiesSet(); client.setTcpSocketFactorySupport(clientTcpSocketFactorySupport); client.start(); @@ -345,19 +365,22 @@ Certificate fingerprints: DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks", "test.truststore.ks", "secret", "secret"); sslContextSupport.setProtocol("SSL"); - DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(sslContextSupport); + DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport = + new DefaultTcpNioSSLConnectionSupport(sslContextSupport); tcpNioConnectionSupport.afterPropertiesSet(); server.setTcpNioConnectionSupport(tcpNioConnectionSupport); final List> messages = new ArrayList>(); final CountDownLatch latch = new CountDownLatch(1); server.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { - System.out.println("Server" + message); +// System.out.println("Server" + message); messages.add(message); latch.countDown(); return false; } + }); server.setMapper(new SSLMapper()); server.start(); @@ -366,11 +389,13 @@ Certificate fingerprints: TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", server.getPort()); client.setTcpNioConnectionSupport(tcpNioConnectionSupport); client.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { - System.out.println("Client" + message); +// System.out.println("Client" + message); return false; } + }); client.start(); @@ -387,7 +412,8 @@ Certificate fingerprints: TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0); TcpSSLContextSupport serverSslContextSupport = new DefaultTcpSSLContextSupport("server.ks", "server.truststore.ks", "secret", "secret"); - DefaultTcpNioSSLConnectionSupport serverTcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport); + DefaultTcpNioSSLConnectionSupport serverTcpNioConnectionSupport = + new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport); serverTcpNioConnectionSupport.afterPropertiesSet(); server.setTcpNioConnectionSupport(serverTcpNioConnectionSupport); final List> messages = new ArrayList>(); @@ -395,18 +421,21 @@ Certificate fingerprints: final Replier replier = new Replier(); server.registerSender(replier); server.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { - System.out.println("Server:" + message); +// System.out.println("Server:" + message); messages.add(message); try { replier.send(message); - } catch (Exception e) { + } + catch (Exception e) { e.printStackTrace(); } latch.countDown(); return false; } + }); ByteArrayCrLfSerializer deserializer = new ByteArrayCrLfSerializer(); deserializer.setMaxMessageSize(120000); @@ -417,17 +446,20 @@ Certificate fingerprints: TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", server.getPort()); TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks", "client.truststore.ks", "secret", "secret"); - DefaultTcpNioSSLConnectionSupport clientTcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(clientSslContextSupport); + DefaultTcpNioSSLConnectionSupport clientTcpNioConnectionSupport = + new DefaultTcpNioSSLConnectionSupport(clientSslContextSupport); clientTcpNioConnectionSupport.afterPropertiesSet(); client.setTcpNioConnectionSupport(clientTcpNioConnectionSupport); client.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { - System.out.println("Client:" + message); +// System.out.println("Client:" + message); messages.add(message); latch.countDown(); return false; } + }); client.setDeserializer(deserializer); client.start(); @@ -464,6 +496,7 @@ Certificate fingerprints: sslEngine.beginHandshake(); this.connection.send(message); } + } private static class SSLMapper extends TcpMessageMapper { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 6b7f10db7c..3a6b9b4e4c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -37,7 +37,6 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSe import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.ip.util.TestingUtilities; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.messaging.Message; /** @@ -48,14 +47,14 @@ public class TcpNioConnectionReadTests { private final CountDownLatch latch = new CountDownLatch(1); - private AbstractServerConnectionFactory getConnectionFactory(int port, + private AbstractServerConnectionFactory getConnectionFactory( AbstractByteArraySerializer serializer, TcpListener listener) throws Exception { - return getConnectionFactory(port, serializer, listener, null); + return getConnectionFactory(serializer, listener, null); } - private AbstractServerConnectionFactory getConnectionFactory(int port, + private AbstractServerConnectionFactory getConnectionFactory( AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) throws Exception { - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); scf.setSerializer(serializer); scf.setDeserializer(serializer); scf.registerListener(listener); @@ -69,22 +68,23 @@ public class TcpNioConnectionReadTests { @Test public void testReadLength() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { responses.add(message); semaphore.release(); return false; } + }); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendLength(port, latch); + CountDownLatch done = SocketTestUtils.testSendLength(scf.getPort(), latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); @@ -102,11 +102,11 @@ public class TcpNioConnectionReadTests { @SuppressWarnings("unchecked") @Test public void testFragmented() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { responses.add(message); @@ -119,12 +119,13 @@ public class TcpNioConnectionReadTests { semaphore.release(); return false; } + }); int howMany = 2; scf.setBacklog(howMany + 5); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendFragmented(port, howMany, false); + CountDownLatch done = SocketTestUtils.testSendFragmented(scf.getPort(), howMany, false); assertTrue(semaphore.tryAcquire(howMany, 20000, TimeUnit.MILLISECONDS)); assertEquals("Expected", howMany, responses.size()); for (int i = 0; i < howMany; i++) { @@ -138,22 +139,23 @@ public class TcpNioConnectionReadTests { @SuppressWarnings("unchecked") @Test public void testReadStxEtx() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { responses.add(message); semaphore.release(); return false; } + }); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendStxEtx(port, latch); + CountDownLatch done = SocketTestUtils.testSendStxEtx(scf.getPort(), latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); @@ -169,22 +171,23 @@ public class TcpNioConnectionReadTests { @SuppressWarnings("unchecked") @Test public void testReadCrLf() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { responses.add(message); semaphore.release(); return false; } + }); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendCrLf(port, latch); + CountDownLatch done = SocketTestUtils.testSendCrLf(scf.getPort(), latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); @@ -199,33 +202,37 @@ public class TcpNioConnectionReadTests { @Test public void testReadLengthOverflow() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { semaphore.release(); return false; } + }, new TcpSender() { + @Override public void addNewConnection(TcpConnection connection) { added.add(connection); semaphore.release(); } + @Override public void removeDeadConnection(TcpConnection connection) { removed.add(connection); semaphore.release(); } + }); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendLengthOverflow(port); + CountDownLatch done = SocketTestUtils.testSendLengthOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); @@ -236,34 +243,38 @@ public class TcpNioConnectionReadTests { @Test public void testReadStxEtxOverflow() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); serializer.setMaxMessageSize(1024); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { semaphore.release(); return false; } + }, new TcpSender() { + @Override public void addNewConnection(TcpConnection connection) { added.add(connection); semaphore.release(); } + @Override public void removeDeadConnection(TcpConnection connection) { removed.add(connection); semaphore.release(); } + }); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(port); + CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); @@ -274,34 +285,38 @@ public class TcpNioConnectionReadTests { @Test public void testReadCrLfOverflow() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); serializer.setMaxMessageSize(1024); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { semaphore.release(); return false; } + }, new TcpSender() { + @Override public void addNewConnection(TcpConnection connection) { added.add(connection); semaphore.release(); } + @Override public void removeDeadConnection(TcpConnection connection) { removed.add(connection); semaphore.release(); } + }); // Fire up the sender. - CountDownLatch done = SocketTestUtils.testSendCrLfOverflow(port); + CountDownLatch done = SocketTestUtils.testSendCrLfOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); @@ -317,31 +332,35 @@ public class TcpNioConnectionReadTests { */ @Test public void testCloseCleanupNoData() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); serializer.setMaxMessageSize(1024); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { semaphore.release(); return false; } + }, new TcpSender() { + @Override public void addNewConnection(TcpConnection connection) { added.add(connection); semaphore.release(); } + @Override public void removeDeadConnection(TcpConnection connection) { removed.add(connection); semaphore.release(); } + }); - Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + Socket socket = SocketFactory.getDefault().createSocket("localhost", scf.getPort()); socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); @@ -357,31 +376,35 @@ public class TcpNioConnectionReadTests { */ @Test public void testCloseCleanupPartialData() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); serializer.setMaxMessageSize(1024); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { + @Override public boolean onMessage(Message message) { semaphore.release(); return false; } + }, new TcpSender() { + @Override public void addNewConnection(TcpConnection connection) { added.add(connection); semaphore.release(); } + @Override public void removeDeadConnection(TcpConnection connection) { removed.add(connection); semaphore.release(); } + }); - Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + Socket socket = SocketFactory.getDefault().createSocket("localhost", scf.getPort()); socket.getOutputStream().write("partial".getBytes()); socket.close(); whileOpen(semaphore, added); @@ -428,12 +451,11 @@ public class TcpNioConnectionReadTests { private void testClosureMidMessageGuts(AbstractByteArraySerializer serializer, String shortMessage) throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { @Override public boolean onMessage(Message message) { responses.add(message); @@ -451,7 +473,7 @@ public class TcpNioConnectionReadTests { semaphore.release(); } }); - Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + Socket socket = SocketFactory.getDefault().createSocket("localhost", scf.getPort()); socket.getOutputStream().write(shortMessage.getBytes()); socket.close(); whileOpen(semaphore, added); 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 c105c1d1a6..2a18f75bed 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 @@ -83,7 +83,6 @@ import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.MapMessageConverter; import org.springframework.integration.test.rule.Log4jLevelAdjuster; -import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.CompositeExecutor; import org.springframework.messaging.Message; @@ -115,11 +114,6 @@ public class TcpNioConnectionTests { @Test public void testWriteTimeout() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port); - factory.setApplicationEventPublisher(nullPublisher); - factory.setSoTimeout(1000); - factory.start(); final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch done = new CountDownLatch(1); final AtomicReference serverSocket = new AtomicReference(); @@ -128,8 +122,8 @@ public class TcpNioConnectionTests { @SuppressWarnings("unused") public void run() { try { - logger.debug(testName.getMethodName() + " starting server for " + port); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort()); serverSocket.set(server); latch.countDown(); Socket s = server.accept(); @@ -142,6 +136,11 @@ public class TcpNioConnectionTests { } }); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); + factory.setApplicationEventPublisher(nullPublisher); + factory.setSoTimeout(1000); + factory.start(); try { TcpConnection connection = factory.getConnection(); connection.send(MessageBuilder.withPayload(new byte[1000000]).build()); @@ -156,11 +155,6 @@ public class TcpNioConnectionTests { @Test public void testReadTimeout() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port); - factory.setApplicationEventPublisher(nullPublisher); - factory.setSoTimeout(1000); - factory.start(); final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch done = new CountDownLatch(1); final AtomicReference serverSocket = new AtomicReference(); @@ -168,8 +162,8 @@ public class TcpNioConnectionTests { @Override public void run() { try { - logger.debug(testName.getMethodName() + " starting server for " + port); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort()); serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); @@ -184,6 +178,11 @@ public class TcpNioConnectionTests { } }); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); + factory.setApplicationEventPublisher(nullPublisher); + factory.setSoTimeout(1000); + factory.start(); try { TcpConnection connection = factory.getConnection(); connection.send(MessageBuilder.withPayload("Test").build()); @@ -205,19 +204,14 @@ public class TcpNioConnectionTests { @Test public void testMemoryLeak() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port); - factory.setApplicationEventPublisher(nullPublisher); - factory.setNioHarvestInterval(100); - factory.start(); final CountDownLatch latch = new CountDownLatch(1); final AtomicReference serverSocket = new AtomicReference(); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override public void run() { try { - logger.debug(testName.getMethodName() + " starting server for " + port); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort()); serverSocket.set(server); latch.countDown(); Socket socket = server.accept(); @@ -230,6 +224,11 @@ public class TcpNioConnectionTests { } }); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); + factory.setApplicationEventPublisher(nullPublisher); + factory.setNioHarvestInterval(100); + factory.start(); try { TcpConnection connection = factory.getConnection(); Map connections = factory.getConnections(); @@ -586,8 +585,7 @@ public class TcpNioConnectionTests { @Test public void testAssemblerUsesSecondaryExecutor() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(0); factory.setApplicationEventPublisher(nullPublisher); CompositeExecutor compositeExec = compositeExecutor(); @@ -609,6 +607,8 @@ public class TcpNioConnectionTests { }); factory.start(); + TestingUtilities.waitListening(factory, null); + int port = factory.getPort(); Socket socket = null; int n = 0; @@ -633,8 +633,7 @@ public class TcpNioConnectionTests { @Test public void testAllMessagesDelivered() throws Exception { final int numberOfSockets = 100; - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(0); factory.setApplicationEventPublisher(nullPublisher); CompositeExecutor compositeExec = compositeExecutor(); @@ -653,6 +652,8 @@ public class TcpNioConnectionTests { }); factory.start(); + TestingUtilities.waitListening(factory, null); + int port = factory.getPort(); Socket[] sockets = new Socket[numberOfSockets]; for (int i = 0; i < numberOfSockets; i++) { @@ -721,8 +722,7 @@ public class TcpNioConnectionTests { @Test public void int3453RaceTest() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); - TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(port); + TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(0); final CountDownLatch connectionLatch = new CountDownLatch(1); factory.setApplicationEventPublisher(new ApplicationEventPublisher() { @@ -759,6 +759,7 @@ public class TcpNioConnectionTests { factory.setTaskExecutor(te); factory.start(); TestingUtilities.waitListening(factory, 10000L); + int port = factory.getPort(); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); assertTrue(connectionLatch.await(10, TimeUnit.SECONDS)); 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 90cdcf63a2..fac442025d 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-2013 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,7 +35,6 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer; import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.util.SocketUtils; /** * @author Gary Russell @@ -56,10 +55,9 @@ public class TcpNioConnectionWriteTests { @Test public void testWriteLengthHeader() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault() - .createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -93,10 +91,9 @@ public class TcpNioConnectionWriteTests { @Test public void testWriteStxEtx() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault() - .createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -130,10 +127,9 @@ public class TcpNioConnectionWriteTests { @Test public void testWriteCrLf() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault() - .createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -167,10 +163,9 @@ public class TcpNioConnectionWriteTests { @Test public void testWriteLengthHeaderDirect() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault() - .createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -204,10 +199,9 @@ public class TcpNioConnectionWriteTests { @Test public void testWriteStxEtxDirect() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault() - .createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -241,10 +235,9 @@ public class TcpNioConnectionWriteTests { @Test public void testWriteCrLfDirect() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault() - .createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { 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 50c41434d0..de51c64dff 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 @@ -38,7 +38,6 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.integration.ip.util.SocketTestUtils; -import org.springframework.integration.test.util.SocketUtils; /** * @author Gary Russell @@ -49,8 +48,8 @@ public class DeserializationTests { @Test public void testReadLength() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendLength(port, null); Socket socket = server.accept(); @@ -68,8 +67,8 @@ public class DeserializationTests { @Test public void testReadStxEtx() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendStxEtx(port, null); Socket socket = server.accept(); @@ -87,8 +86,8 @@ public class DeserializationTests { @Test public void testReadCrLf() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendCrLf(port, null); Socket socket = server.accept(); @@ -106,8 +105,8 @@ public class DeserializationTests { @Test public void testReadRaw() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); SocketTestUtils.testSendRaw(port); Socket socket = server.accept(); @@ -121,8 +120,8 @@ public class DeserializationTests { @Test public void testReadSerialized() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendSerialized(port); Socket socket = server.accept(); @@ -138,8 +137,8 @@ public class DeserializationTests { @Test public void testReadLengthOverflow() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendLengthOverflow(port); Socket socket = server.accept(); @@ -160,8 +159,8 @@ public class DeserializationTests { @Test public void testReadStxEtxTimeout() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(port); Socket socket = server.accept(); @@ -182,8 +181,8 @@ public class DeserializationTests { @Test public void testReadStxEtxOverflow() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(port); Socket socket = server.accept(); @@ -205,8 +204,8 @@ public class DeserializationTests { @Test public void testReadCrLfTimeout() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch latch = SocketTestUtils.testSendCrLfOverflow(port); Socket socket = server.accept(); @@ -227,8 +226,8 @@ public class DeserializationTests { @Test public void testReadCrLfOverflow() throws Exception { - int port = SocketUtils.findAvailableServerSocket(); - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + int port = server.getLocalPort(); server.setSoTimeout(10000); CountDownLatch latch = SocketTestUtils.testSendCrLfOverflow(port); Socket socket = server.accept(); @@ -305,9 +304,9 @@ public class DeserializationTests { @Override public void publishEvent(Object event) { - + } - + } Publisher publisher = new Publisher(); ByteArrayInputStream bais = new ByteArrayInputStream(data); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java index b51e1e38be..615672db38 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -33,7 +33,6 @@ import javax.net.SocketFactory; import org.junit.Test; import org.springframework.core.serializer.DefaultSerializer; -import org.springframework.integration.test.util.SocketUtils; /** * @author Gary Russell @@ -43,9 +42,9 @@ public class SerializationTests { @Test public void testWriteLengthHeader() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -80,9 +79,9 @@ public class SerializationTests { @Test public void testWriteStxEtx() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -117,9 +116,9 @@ public class SerializationTests { @Test public void testWriteCrLf() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -154,9 +153,9 @@ public class SerializationTests { @Test public void testWriteRaw() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -191,9 +190,9 @@ public class SerializationTests { @Test public void testWriteSerialized() throws Exception { - final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); server.setSoTimeout(10000); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() {