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

INT-2903 Fix Timing Problem in TCP Timeout Tests

Browse Source 
Use latches and the new CLOSE event instead of arbitrary waits for
sockets to close.

Factor out common test code into setup methods.

removed system.out
This commit is contained in:
Gary Russell
2013-04-30 14:56:44 -04:00
committed by Mark Fisher
parent e24efc89b3
commit b654044fd2
1 changed files with 81 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
@@ -22,10 +23,16 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent.TcpConnectionEventType;
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
@@ -34,7 +41,6 @@ import org.springframework.integration.test.util.TestUtils;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class ConnectionTimeoutTests {
@@ -42,23 +48,8 @@ public class ConnectionTimeoutTests {
public void testDefaultTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
server.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
return false;
}
});
TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
client.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
}
public void removeDeadConnection(TcpConnection connection) {
}
});
client.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
return false;
}
});
this.setupCallbacks(server, client, 0);
server.start();
TestingUtilities.waitListening(server, null);
client.start();
@@ -75,19 +66,10 @@ public class ConnectionTimeoutTests {
public void testNetSimpleTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
server.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
return false;
}
});
TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
client.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
}
public void removeDeadConnection(TcpConnection connection) {
}
});
this.setupCallbacks(server, client, 0);
client.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
@@ -95,11 +77,12 @@ public class ConnectionTimeoutTests {
client.setSoTimeout(1000);
server.start();
TestingUtilities.waitListening(server, null);
CountDownLatch clientCloseLatch = getCloseLatch(client);
client.start();
TcpConnection connection = client.getConnection();
Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
assertEquals(1000, socket.getSoTimeout());
Thread.sleep(1100);
assertTrue(clientCloseLatch.await(3, TimeUnit.SECONDS));
assertFalse(connection.isOpen());
server.stop();
client.stop();
@@ -114,55 +97,8 @@ public class ConnectionTimeoutTests {
public void testNetReplyNotTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
server.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
try {
Thread.sleep(1200);
serverConnection.get().send(message);
}
catch (Exception e) {
e.printStackTrace();
}
return false;
}
});
server.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
serverConnection.set(connection);
}
public void removeDeadConnection(TcpConnection connection) {
}
});
TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
client.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
}
public void removeDeadConnection(TcpConnection connection) {
}
});
final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
client.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
reply.set(message);
return false;
}
});
client.setSoTimeout(2000);
server.start();
TestingUtilities.waitListening(server, null);
client.start();
TcpConnection connection = client.getConnection();
Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
assertEquals(2000, socket.getSoTimeout());
Thread.sleep(1000);
connection.send(MessageBuilder.withPayload("foo").build());
Thread.sleep(1400);
assertNotNull(reply.get());
Thread.sleep(2200);
assertFalse(connection.isOpen());
server.stop();
client.stop();
this.notTimeoutGuts(server, client);
}
/**
@@ -174,50 +110,34 @@ public class ConnectionTimeoutTests {
public void testNioReplyNotTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
server.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
try {
Thread.sleep(1200);
serverConnection.get().send(message);
}
catch (Exception e) {
e.printStackTrace();
}
return false;
}
});
server.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
serverConnection.set(connection);
}
public void removeDeadConnection(TcpConnection connection) {
}
});
TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
client.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
}
public void removeDeadConnection(TcpConnection connection) {
}
});
this.notTimeoutGuts(server, client);
}
private void notTimeoutGuts(AbstractServerConnectionFactory server, AbstractClientConnectionFactory client)
throws Exception, InterruptedException {
this.setupCallbacks(server, client, 1200);
final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
final CountDownLatch replyLatch = new CountDownLatch(1);
client.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
reply.set(message);
replyLatch.countDown();
return false;
}
});
client.setSoTimeout(2000);
server.start();
TestingUtilities.waitListening(server, null);
CountDownLatch clientClosedLatch = getCloseLatch(client);
client.start();
TcpConnection connection = client.getConnection();
Thread.sleep(1000);
connection.send(MessageBuilder.withPayload("foo").build());
Thread.sleep(1400);
assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
assertNotNull(reply.get());
Thread.sleep(4200);
assertTrue(clientClosedLatch.await(10, TimeUnit.SECONDS));
assertFalse(connection.isOpen());
server.stop();
client.stop();
@@ -232,35 +152,11 @@ public class ConnectionTimeoutTests {
public void testNetReplyTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
server.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
try {
Thread.sleep(4200);
serverConnection.get().send(message);
}
catch (Exception e) {
e.printStackTrace();
}
return false;
}
});
server.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
serverConnection.set(connection);
}
public void removeDeadConnection(TcpConnection connection) {
}
});
TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
client.registerSender(new TcpSender() {
public void addNewConnection(TcpConnection connection) {
}
public void removeDeadConnection(TcpConnection connection) {
}
});
this.setupCallbacks(server, client, 4500);
final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
client.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
reply.set(message);
return false;
@@ -269,6 +165,7 @@ public class ConnectionTimeoutTests {
client.setSoTimeout(2000);
server.start();
TestingUtilities.waitListening(server, null);
CountDownLatch clientCloseLatch = getCloseLatch(client);
client.start();
TcpConnection connection = client.getConnection();
Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
@@ -277,7 +174,7 @@ public class ConnectionTimeoutTests {
connection.send(MessageBuilder.withPayload("foo").build());
Thread.sleep(1400);
assertTrue(connection.isOpen());
Thread.sleep(2000);
assertTrue(clientCloseLatch.await(2000, TimeUnit.SECONDS));
assertNull(reply.get());
assertFalse(connection.isOpen());
server.stop();
@@ -293,11 +190,43 @@ public class ConnectionTimeoutTests {
public void testNioReplyTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
this.setupCallbacks(server, client, 2100);
final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
client.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
reply.set(message);
return false;
}
});
client.setSoTimeout(1000);
server.start();
TestingUtilities.waitListening(server, null);
CountDownLatch clientCloseLatch = getCloseLatch(client);
client.start();
TcpConnection connection = client.getConnection();
Thread.sleep(500);
connection.send(MessageBuilder.withPayload("foo").build());
Thread.sleep(700);
assertTrue(connection.isOpen());
assertTrue(clientCloseLatch.await(2, TimeUnit.SECONDS));
assertNull(reply.get());
assertFalse(connection.isOpen());
server.stop();
client.stop();
}
private void setupCallbacks(AbstractServerConnectionFactory server, AbstractClientConnectionFactory client,
final int serverDelay) {
client.setComponentName("clientFactory");
server.setComponentName("serverFactory");
final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
server.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
try {
Thread.sleep(2100);
Thread.sleep(serverDelay);
serverConnection.get().send(message);
}
catch (Exception e) {
@@ -307,39 +236,37 @@ public class ConnectionTimeoutTests {
}
});
server.registerSender(new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
serverConnection.set(connection);
}
@Override
public void removeDeadConnection(TcpConnection connection) {
}
});
TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
client.registerSender(new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
}
@Override
public void removeDeadConnection(TcpConnection connection) {
}
});
final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
client.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
reply.set(message);
return false;
}
private CountDownLatch getCloseLatch(AbstractClientConnectionFactory client) {
final CountDownLatch clientClosedLatch;
clientClosedLatch = new CountDownLatch(1);
client.setApplicationEventPublisher(new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
TcpConnectionEvent tcpEvent = (TcpConnectionEvent) event;
if (tcpEvent.getType() == TcpConnectionEventType.CLOSE) {
clientClosedLatch.countDown();
}
}
});
client.setSoTimeout(1000);
server.start();
TestingUtilities.waitListening(server, null);
client.start();
TcpConnection connection = client.getConnection();
Thread.sleep(500);
connection.send(MessageBuilder.withPayload("foo").build());
Thread.sleep(700);
assertTrue(connection.isOpen());
Thread.sleep(1000);
assertNull(reply.get());
assertFalse(connection.isOpen());
server.stop();
client.stop();
return clientClosedLatch;
}
}