Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

TCP: Tests: Let OS Choose Test Port

Browse Source 
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
This commit is contained in:
Gary Russell
2015-10-29 15:01:28 -04:00
committed by Artem Bilan
parent e5577b1e50
commit fb766a951c
12 changed files with 561 additions and 394 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java
View File

@@ -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();

51
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java
View File

@@ -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));

134
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -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> serverSocket = new AtomicReference<ServerSocket>();
@@ -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<Socket> sockets = new ArrayList<Socket>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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<String> lastReceived = new AtomicReference<String>();
@@ -690,7 +716,6 @@ public class TcpOutboundGatewayTests {
public void run() {
List<Socket> sockets = new ArrayList<Socket>();
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<Socket> sockets = new ArrayList<Socket>();
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (!done.get()) {
sockets.add(server.accept());

94
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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<Socket> sockets = new LinkedList<Socket>();
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());

302
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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<String> results = new HashSet<String>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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<String> results = new HashSet<String>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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<String> results = new HashSet<String>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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<String> results = new HashSet<String>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
final List<Socket> serverSockets = new ArrayList<Socket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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> serverSocket = new AtomicReference<ServerSocket>();
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());
}
}
}

16
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java
View File

@@ -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");

101
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java
View File

@@ -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<Socket> (){
when(serverSocket.accept()).thenReturn(socket).then(new Answer<Socket>() {
@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<String>("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<Message<?>> messages = new ArrayList<Message<?>>();
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<Message<?>> messages = new ArrayList<Message<?>>();
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<Message<?>> messages = new ArrayList<Message<?>>();
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<Message<?>> messages = new ArrayList<Message<?>>();
@@ -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 {

92
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -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<Message<?>> responses = new ArrayList<Message<?>>();
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<Message<?>> responses = new ArrayList<Message<?>>();
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<Message<?>> responses = new ArrayList<Message<?>>();
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<Message<?>> responses = new ArrayList<Message<?>>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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);

57
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
View File

@@ -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> serverSocket = new AtomicReference<ServerSocket>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
@@ -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> serverSocket = new AtomicReference<ServerSocket>();
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<SocketChannel, TcpNioConnection> 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));

33
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java
View File

@@ -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() {

45
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java
View File

@@ -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);

23
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java
View File

@@ -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() {