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

Merge pull request #797 from garyrussell/INT-2903

Browse Source 
* INT-2903: Fix Timing Problem in TCP Timeout Tests
This commit is contained in:
Mark Fisher
2013-05-07 10:32:19 -04:00
parent e24efc89b3 b654044fd2
commit fd9dc75550
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;
}
}