INT-2620 Fix Race Condition in Failover Tests

Sporadic test failures (fairly consistently on Mac).

Problem was some tests sent a message to a client socket before the
close notification had been received.

Wait for client connection to close rather than waiting for the
server to stop.

Also, with NIO, close the channel on accept exception.

Also improve debugging by including a log of new connections
including the id.

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2011 the original author or authors.
+ * Copyright 2001-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.
@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.Message;
@@ -87,6 +88,9 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 		try {
 			this.soLinger = socket.getSoLinger();
 		} catch (SocketException e) { }
+		if (logger.isDebugEnabled()) {
+			logger.debug("New connection " + this.getConnectionId());
+		}
 	}
 
 	public void afterSend(Message<?> message) throws Exception {

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java

@@ -156,17 +156,23 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 			channel.close();
 		}
 		else {
-			channel.configureBlocking(false);
-			Socket socket = channel.socket();
-			setSocketAttributes(socket);
-			TcpNioConnection connection = createTcpNioConnection(channel);
-			if (connection == null) {
-				return;
+			try {
+				channel.configureBlocking(false);
+				Socket socket = channel.socket();
+				setSocketAttributes(socket);
+				TcpNioConnection connection = createTcpNioConnection(channel);
+				if (connection == null) {
+					return;
+				}
+				connection.setTaskExecutor(this.getTaskExecutor());
+				connection.setLastRead(now);
+				this.channelMap.put(channel, connection);
+				channel.register(selector, SelectionKey.OP_READ, connection);
+			}
+			catch (Exception e) {
+				logger.error("Exception accepting new connection", e);
+				channel.close();
 			}
-			connection.setTaskExecutor(this.getTaskExecutor());
-			connection.setLastRead(now);
-			this.channelMap.put(channel, connection);
-			channel.register(selector, SelectionKey.OP_READ, connection);
 		}
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.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.
@@ -15,7 +15,12 @@
  */
 package org.springframework.integration.ip.util;
 
+import java.util.Collection;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpConnection;
 
 /**
  * Convenience class providing methods for testing IP components.
@@ -88,4 +93,32 @@ public class TestingUtilities {
 		}
 	}
 
+	/**
+	 * Wait for up to 10 seconds for the connection factory to have the specified number
+	 * of connections.
+	 * @param factory The factory.
+	 * @param n The required number of connections.
+	 * @throws Exception IllegalStateException if the count does not match.
+	 */
+	public static void waitUntilFactoryHasThisNumberOfConnections(AbstractConnectionFactory factory, int n)
+			throws Exception {
+		int timer = 0;
+		DirectFieldAccessor accessor = new DirectFieldAccessor(factory);
+		@SuppressWarnings("unchecked")
+		Collection<TcpConnection> connections = (Collection<TcpConnection>) accessor.getPropertyValue("connections");
+		while (timer < 10000) {
+			int open = 0;
+			for (TcpConnection connection : connections) {
+				if (connection.isOpen()) {
+					open++;
+				}
+			}
+			if (open == n) {
+				return;
+			}
+			Thread.sleep(100);
+			timer += 100;
+		}
+		throw new IllegalStateException("Connections=" + connections.size() + "wanted=" + n);
+	}
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.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.
@@ -39,6 +39,7 @@ import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessagingException;
@@ -335,7 +336,7 @@ public class FailoverClientConnectionFactoryTests {
 		Message<?> replyMessage = replyChannel.receive(10000);
 		assertNotNull(replyMessage);
 		server1.stop();
-		TestingUtilities.waitStopListening(server1, null);
+		TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(client1, 0);
 		outGateway.handleMessage(message);
 		socket = getSocket(client2);
 		port2 = socket.getLocalPort();
@@ -343,6 +344,7 @@ public class FailoverClientConnectionFactoryTests {
 		replyMessage = replyChannel.receive(10000);
 		assertNotNull(replyMessage);
 		gateway2.stop();
+		outGateway.stop();
 	}
 
 	private Socket getSocket(AbstractClientConnectionFactory client) throws Exception {