From b654044fd2da7f6a794d24f173456257aaacf8b4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 30 Apr 2013 14:56:44 -0400 Subject: [PATCH] INT-2903 Fix Timing Problem in TCP Timeout Tests 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 --- .../connection/ConnectionTimeoutTests.java | 235 ++++++------------ 1 file changed, 81 insertions(+), 154 deletions(-) diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java index c379bcc474..4c5052aac7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java @@ -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 serverConnection = new AtomicReference(); - 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> reply = new AtomicReference>(); - 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 serverConnection = new AtomicReference(); - 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> reply = new AtomicReference>(); + 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 serverConnection = new AtomicReference(); - 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> reply = new AtomicReference>(); 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> reply = new AtomicReference>(); + 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 serverConnection = new AtomicReference(); 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> reply = new AtomicReference>(); - 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; } + }