INT-2459 Add Support for TCP Failover

Implemented a ConnectionFactory that wraps a list
of connection factories, used to fail over if
a connection fails.

INT-2459 Polishing

Rebase; fix test; move boolean resets to finally block
in server factories.

INT-2459 Polishing

PR Review

Fix a test that could block indefinitely when error
in code under test.

spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -18,7 +18,7 @@ package org.springframework.integration.ip;
 
 /**
  * Headers for Messages mapped from IP datagram packets.
- * 
+ *
  * @author Mark Fisher
  * @author Gary Russell
  * @author Dave Syer
@@ -32,15 +32,15 @@ public abstract class IpHeaders {
 
 	public static final String HOSTNAME = IP + "hostname";
 
-	public static final String IP_ADDRESS = IP + "address"; 
+	public static final String IP_ADDRESS = IP + "address";
 
 	public static final String ACK_ADDRESS = IP + "ackTo";
 
 	public static final String ACK_ID = IP + "ackId";
 
-	public static final String REMOTE_PORT = TCP + "remote_port";
+	public static final String REMOTE_PORT = TCP + "remotePort";
 
-	public static final String CONNECTION_ID = IP + "connection_id";
+	public static final String CONNECTION_ID = IP + "connectionId";
 
 	/**
 	 * Use apply-sequence and sequenceNumber instead
@@ -49,6 +49,8 @@ public abstract class IpHeaders {
 	@Deprecated
 	public static final String CONNECTION_SEQ = IP + "connection_seq";
 
+	public static final String ACTUAL_CONNECTION_ID = IP + "actualConnectionId";
+
 	private IpHeaders() {}
 
 }

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

@@ -37,7 +37,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	public AbstractClientConnectionFactory(String host, int port) {
 		super(host, port);
 	}
-	
+
 	/**
 	 * Obtains a connection - if {@link #setSingleUse(boolean)} was called with
 	 * true, a new connection is returned; otherwise a single connection is
@@ -46,26 +46,26 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	public TcpConnection getConnection() throws Exception {
 		this.checkActive();
 		if (this.isSingleUse()) {
-			return getOrMakeConnection();
+			return obtainConnection();
 		} else {
 			synchronized(this) {
-				TcpConnection connection = getOrMakeConnection();
+				TcpConnection connection = obtainConnection();
 				this.setTheConnection(connection);
 				return connection;
 			}
 		}
 	}
 
-	protected abstract TcpConnection getOrMakeConnection() throws Exception;
+	protected abstract TcpConnection obtainConnection() throws Exception;
 
 	/**
 	 * Transfers attributes such as (de)serializers, singleUse etc to a new connection.
-	 * When the connection factory has a reference to a TCPListener (to read 
+	 * When the connection factory has a reference to a TCPListener (to read
 	 * responses), or for single use connections, the connection is executed.
-	 * Single use connections need to read from the connection in order to 
+	 * Single use connections need to read from the connection in order to
 	 * close it after the socket timeout.
 	 * @param connection The new connection.
-	 * @param socket The new socket. 
+	 * @param socket The new socket.
 	 */
 	protected void initializeConnection(TcpConnection connection, Socket socket) {
 		TcpListener listener = this.getListener();

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

@@ -54,7 +54,7 @@ import org.springframework.util.Assert;
  *
  */
 public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
-		implements ConnectionFactory, Runnable, SmartLifecycle, OrderlyShutdownCapable {
+		implements ConnectionFactory, SmartLifecycle, OrderlyShutdownCapable {
 
 	protected static final int DEFAULT_REPLY_TIMEOUT = 10000;
 
@@ -421,16 +421,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 	 */
 	public abstract void close();
 
-	/**
-	 * Starts the listening process.
-	 */
 	public void start() {
-		synchronized (this.lifecycleMonitor) {
-			if (!this.active) {
-				this.active = true;
-				this.getTaskExecutor().execute(this);
-			}
-		}
 		if (logger.isInfoEnabled()) {
 			logger.info("started " + this);
 		}

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

@@ -30,7 +30,8 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @since 2.0
  */
-public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
+public abstract class AbstractServerConnectionFactory
+		extends AbstractConnectionFactory implements Runnable {
 
 	private static final int DEFAULT_BACKLOG = 5;
 
@@ -49,6 +50,16 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 		super(port);
 	}
 
+	@Override
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.isActive()) {
+				this.setActive(true);
+				this.getTaskExecutor().execute(this);
+			}
+		}
+		super.start();
+	}
 
 	/**
 	 * Not supported because the factory manages multiple connections and this

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

@@ -66,10 +66,12 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 		this.pool.setWaitTimeout(connectionWaitTimeout);
 	}
 
+	@Override
 	public synchronized void setPoolSize(int poolSize) {
 		this.pool.setPoolSize(poolSize);
 	}
 
+	@Override
 	public int getPoolSize() {
 		return this.pool.getPoolSize();
 	}
@@ -86,7 +88,8 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 		return this.pool.getAllocatedCount();
 	}
 
-	public TcpConnection getOrMakeConnection() throws Exception {
+	@Override
+	public TcpConnection obtainConnection() throws Exception {
 		return new CachedConnection(this.pool.getItem());
 	}
 
@@ -139,10 +142,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 
 ///////////////// DELEGATE METHODS ///////////////////////
 
-	public void run() {
-
-	}
-
+	@Override
 	public boolean isRunning() {
 		return targetConnectionFactory.isRunning();
 	}
@@ -152,170 +152,212 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 		targetConnectionFactory.close();
 	}
 
+	@Override
 	public int hashCode() {
 		return targetConnectionFactory.hashCode();
 	}
 
+	@Override
 	public void setComponentName(String componentName) {
 		targetConnectionFactory.setComponentName(componentName);
 	}
 
+	@Override
 	public String getComponentType() {
 		return targetConnectionFactory.getComponentType();
 	}
 
+	@Override
 	public boolean equals(Object obj) {
 		return targetConnectionFactory.equals(obj);
 	}
 
+	@Override
 	public int getSoTimeout() {
 		return targetConnectionFactory.getSoTimeout();
 	}
 
+	@Override
 	public void setSoTimeout(int soTimeout) {
 		targetConnectionFactory.setSoTimeout(soTimeout);
 	}
 
+	@Override
 	public int getSoReceiveBufferSize() {
 		return targetConnectionFactory.getSoReceiveBufferSize();
 	}
 
+	@Override
 	public void setSoReceiveBufferSize(int soReceiveBufferSize) {
 		targetConnectionFactory.setSoReceiveBufferSize(soReceiveBufferSize);
 	}
 
+	@Override
 	public int getSoSendBufferSize() {
 		return targetConnectionFactory.getSoSendBufferSize();
 	}
 
+	@Override
 	public void setSoSendBufferSize(int soSendBufferSize) {
 		targetConnectionFactory.setSoSendBufferSize(soSendBufferSize);
 	}
 
+	@Override
 	public boolean isSoTcpNoDelay() {
 		return targetConnectionFactory.isSoTcpNoDelay();
 	}
 
+	@Override
 	public void setSoTcpNoDelay(boolean soTcpNoDelay) {
 		targetConnectionFactory.setSoTcpNoDelay(soTcpNoDelay);
 	}
 
+	@Override
 	public int getSoLinger() {
 		return targetConnectionFactory.getSoLinger();
 	}
 
+	@Override
 	public void setSoLinger(int soLinger) {
 		targetConnectionFactory.setSoLinger(soLinger);
 	}
 
+	@Override
 	public boolean isSoKeepAlive() {
 		return targetConnectionFactory.isSoKeepAlive();
 	}
 
+	@Override
 	public void setSoKeepAlive(boolean soKeepAlive) {
 		targetConnectionFactory.setSoKeepAlive(soKeepAlive);
 	}
 
+	@Override
 	public int getSoTrafficClass() {
 		return targetConnectionFactory.getSoTrafficClass();
 	}
 
+	@Override
 	public void setSoTrafficClass(int soTrafficClass) {
 		targetConnectionFactory.setSoTrafficClass(soTrafficClass);
 	}
 
+	@Override
 	public String getHost() {
 		return targetConnectionFactory.getHost();
 	}
 
+	@Override
 	public int getPort() {
 		return targetConnectionFactory.getPort();
 	}
 
+	@Override
 	public TcpListener getListener() {
 		return targetConnectionFactory.getListener();
 	}
 
+	@Override
 	public TcpSender getSender() {
 		return targetConnectionFactory.getSender();
 	}
 
+	@Override
 	public Serializer<?> getSerializer() {
 		return targetConnectionFactory.getSerializer();
 	}
 
+	@Override
 	public Deserializer<?> getDeserializer() {
 		return targetConnectionFactory.getDeserializer();
 	}
 
+	@Override
 	public TcpMessageMapper getMapper() {
 		return targetConnectionFactory.getMapper();
 	}
 
+	@Override
 	public void registerListener(TcpListener listener) {
 		targetConnectionFactory.registerListener(listener);
 	}
 
+	@Override
 	public void registerSender(TcpSender sender) {
 		targetConnectionFactory.registerSender(sender);
 	}
 
+	@Override
 	public void setTaskExecutor(Executor taskExecutor) {
 		targetConnectionFactory.setTaskExecutor(taskExecutor);
 	}
 
+	@Override
 	public void setDeserializer(Deserializer<?> deserializer) {
 		targetConnectionFactory.setDeserializer(deserializer);
 	}
 
+	@Override
 	public void setSerializer(Serializer<?> serializer) {
 		targetConnectionFactory.setSerializer(serializer);
 	}
 
+	@Override
 	public void setMapper(TcpMessageMapper mapper) {
 		targetConnectionFactory.setMapper(mapper);
 	}
 
+	@Override
 	public boolean isSingleUse() {
 		return targetConnectionFactory.isSingleUse();
 	}
 
+	@Override
 	public void setSingleUse(boolean singleUse) {
 		targetConnectionFactory.setSingleUse(singleUse);
 	}
 
+	@Override
 	public void setInterceptorFactoryChain(
 			TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
 		targetConnectionFactory
 				.setInterceptorFactoryChain(interceptorFactoryChain);
 	}
 
+	@Override
 	public void setLookupHost(boolean lookupHost) {
 		targetConnectionFactory.setLookupHost(lookupHost);
 	}
 
+	@Override
 	public boolean isLookupHost() {
 		return targetConnectionFactory.isLookupHost();
 	}
 
+	@Override
 	public void start() {
 		this.setActive(true);
 		targetConnectionFactory.start();
+		super.start();
 	}
 
+	@Override
 	public synchronized void stop() {
 		targetConnectionFactory.stop();
 		this.pool.removeAllIdleItems();
 	}
 
+	@Override
 	public int getPhase() {
 		return targetConnectionFactory.getPhase();
 	}
 
+	@Override
 	public boolean isAutoStartup() {
 		return targetConnectionFactory.isAutoStartup();
 	}
 
+	@Override
 	public void stop(Runnable callback) {
 		targetConnectionFactory.stop(callback);
 	}

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

@@ -0,0 +1,345 @@
+/*
+ * Copyright 2002-2012 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp.connection;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+
+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;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Given a list of connection factories, serves up {@link TcpConnection}s
+ * that can iterate over a connection from each factory until the write
+ * succeeds or the list is exhausted.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class FailoverClientConnectionFactory extends AbstractClientConnectionFactory {
+
+	private static final Log logger = LogFactory.getLog(FailoverClientConnectionFactory.class);
+
+	private final List<AbstractClientConnectionFactory> factories;
+
+	public FailoverClientConnectionFactory(List<AbstractClientConnectionFactory> factories) {
+		super("", 0);
+		Assert.notEmpty(factories, "At least one factory is required");
+		this.factories = factories;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		for (AbstractClientConnectionFactory factory : factories) {
+			Assert.state(!(this.isSingleUse() ^ factory.isSingleUse()),
+				"Inconsistent singleUse - delegate factories must match this one");
+		}
+	}
+
+	/**
+	 * Delegate TCP Client Connection factories that are used to receive
+	 * data need a Listener to send the messages to.
+	 * This applies to client factories used for outbound gateways
+	 * or for a pair of collaborating channel adapters.
+	 * <p/>
+	 * During initialization, if a factory detects it has no listener
+	 * it's listening logic (active thread) is terminated.
+	 * <p/>
+	 * The listener registered with a factory is provided to each
+	 * connection it creates so it can call the onMessage() method.
+	 * <p/>
+	 * This code satisfies the first requirement in that this
+	 * listener signals to the factory that it needs to run
+	 * its listening logic.
+	 * <p/>
+	 * When we wrap actual connections with FailoverTcpConnections,
+	 * the connection is given the wrapper as a listener, so it
+	 * can enhance the headers in onMessage(); the wrapper then invokes
+	 * the real listener supplied here, with the modified message.
+	 */
+	@Override
+	public void registerListener(TcpListener listener) {
+		super.registerListener(listener);
+		for (AbstractClientConnectionFactory factory : this.factories) {
+			factory.registerListener(new TcpListener() {
+				public boolean onMessage(Message<?> message) {
+					throw new UnsupportedOperationException("This should never be called");
+				}
+			});
+		}
+	}
+
+	@Override
+	public void registerSender(TcpSender sender) {
+		for (AbstractClientConnectionFactory factory : this.factories) {
+			factory.registerSender(sender);
+		}
+	}
+
+	@Override
+	protected TcpConnection obtainConnection() throws Exception {
+		TcpConnection connection = this.getTheConnection();
+		if (connection != null && connection.isOpen()) {
+			return connection;
+		}
+		return new FailoverTcpConnection(this.factories);
+	}
+
+	@Override
+	public void close() {
+		for (AbstractClientConnectionFactory factory : this.factories) {
+			factory.close();
+		}
+	}
+
+	@Override
+	public void start() {
+		for (AbstractClientConnectionFactory factory : this.factories) {
+			factory.start();
+		}
+		this.setActive(true);
+		super.start();
+	}
+
+	@Override
+	public void stop() {
+		this.setActive(false);
+		for (AbstractClientConnectionFactory factory : this.factories) {
+			factory.stop();
+		}
+	}
+
+	/**
+	 * Returns true if all factories are running
+	 */
+	@Override
+	public boolean isRunning() {
+		boolean isRunning = true;
+		for (AbstractClientConnectionFactory factory : this.factories) {
+			isRunning = !isRunning ? false : factory.isRunning();
+		}
+		return isRunning;
+	}
+
+	/**
+	 * Wrapper for a list of factories; delegates to a connection from
+	 * one of those factories and fails over to another if necessary.
+	 * @author Gary Russell
+	 * @since 2.2
+	 *
+	 */
+	private class FailoverTcpConnection implements TcpConnection, TcpListener {
+
+		private final List<AbstractClientConnectionFactory> factories;
+
+		private final String connectionId;
+
+		private volatile Iterator<AbstractClientConnectionFactory> factoryIterator;
+
+		private volatile AbstractClientConnectionFactory currentFactory;
+
+		private volatile TcpConnection delegate;
+
+		private volatile boolean open = true;
+
+		public FailoverTcpConnection(List<AbstractClientConnectionFactory> factories) throws Exception {
+			this.factories = factories;
+			this.factoryIterator = factories.iterator();
+			findAConnection();
+			this.connectionId = UUID.randomUUID().toString();
+		}
+
+		/**
+		 * Finds a connection from the underlying list of factories. If necessary,
+		 * each factory is tried; including the current one if we wrap around.
+		 * This allows for the condition where the current connection is closed,
+		 * the current factory can serve up a new connection, but all other
+		 * factories are down.
+		 * @throws Exception
+		 */
+		private synchronized void findAConnection() throws Exception {
+			boolean success = false;
+			AbstractClientConnectionFactory lastFactoryToTry = this.currentFactory;
+			AbstractClientConnectionFactory nextFactory = null;
+			if (!this.factoryIterator.hasNext()) {
+				this.factoryIterator = this.factories.iterator();
+			}
+			boolean retried = false;
+			while (!success) {
+				try {
+					nextFactory = this.factoryIterator.next();
+					this.delegate = nextFactory.getConnection();
+					this.delegate.registerListener(this);
+					this.currentFactory = nextFactory;
+					success = this.delegate.isOpen();
+				}
+				catch (IOException e) {
+					if (!this.factoryIterator.hasNext()) {
+						if (retried && lastFactoryToTry == null || lastFactoryToTry == nextFactory) {
+							/*
+							 *  We've tried every factory including the
+							 *  one the current connection was on.
+							 */
+							this.open = false;
+							throw e;
+						}
+						this.factoryIterator = this.factories.iterator();
+						retried = true;
+					}
+				}
+			}
+		}
+
+		public void close() {
+			this.delegate.close();
+			this.open = false;
+		}
+
+		public boolean isOpen() {
+			return this.open;
+		}
+
+		/**
+		 * Sends to the current connection; if it fails, attempts to
+		 * send to a new connection obtained from {@link #findAConnection()}.
+		 * If send fails on a connection from every factory, we give up.
+		 */
+		public synchronized void send(Message<?> message) throws Exception {
+			boolean success = false;
+			AbstractClientConnectionFactory lastFactoryToTry = this.currentFactory;
+			AbstractClientConnectionFactory lastFactoryTried = null;
+			boolean retried = false;
+			while (!success) {
+				try {
+					lastFactoryTried = this.currentFactory;
+					this.delegate.send(message);
+					success = true;
+				}
+				catch (IOException e) {
+					if (retried && lastFactoryTried == lastFactoryToTry) {
+						logger.error("All connection factories exhausted", e);
+						this.open = false;
+						throw e;
+					}
+					retried = true;
+					if (logger.isDebugEnabled()) {
+						logger.debug("Send to " + this.delegate.getConnectionId() + " failed; attempting failover", e);
+					}
+					this.delegate.close();
+					findAConnection();
+					if (logger.isDebugEnabled()) {
+						logger.debug("Failing over to " + this.delegate.getConnectionId());
+					}
+				}
+			}
+		}
+
+		public Object getPayload() throws Exception {
+			return this.delegate.getPayload();
+		}
+
+		public void run() {
+			throw new UnsupportedOperationException("Not supported on FailoverTcpConnection");
+		}
+
+		public String getHostName() {
+			return this.delegate.getHostName();
+		}
+
+		public String getHostAddress() {
+			return this.delegate.getHostAddress();
+		}
+
+		public int getPort() {
+			return this.delegate.getPort();
+		}
+
+		public void registerListener(TcpListener listener) {
+			this.delegate.registerListener(listener);
+		}
+
+		public void registerSender(TcpSender sender) {
+			this.delegate.registerSender(sender);
+		}
+
+		public String getConnectionId() {
+			return this.connectionId;
+		}
+
+		public void setSingleUse(boolean singleUse) {
+			this.delegate.setSingleUse(singleUse);
+		}
+
+		public boolean isSingleUse() {
+			return this.delegate.isSingleUse();
+		}
+
+		public boolean isServer() {
+			return this.delegate.isServer();
+		}
+
+		public void setMapper(TcpMessageMapper mapper) {
+			this.delegate.setMapper(mapper);
+		}
+
+		public Deserializer<?> getDeserializer() {
+			return this.delegate.getDeserializer();
+		}
+
+		public void setDeserializer(Deserializer<?> deserializer) {
+			this.delegate.setDeserializer(deserializer);
+		}
+
+		public Serializer<?> getSerializer() {
+			return this.delegate.getSerializer();
+		}
+
+		public void setSerializer(Serializer<?> serializer) {
+			this.delegate.setSerializer(serializer);
+		}
+
+		public TcpListener getListener() {
+			return this.delegate.getListener();
+		}
+
+		public long incrementAndGetConnectionSequence() {
+			return this.delegate.incrementAndGetConnectionSequence();
+		}
+
+		/**
+		 * We have to intercept the message to replace the connectionId header with
+		 * ours so the listener can correlate a response with a request. We supply
+		 * the actual connectionId in another header for convenience and tracing
+		 * purposes.
+		 */
+		public boolean onMessage(Message<?> message) {
+			return FailoverClientConnectionFactory.this.getListener().onMessage(MessageBuilder.fromMessage(message)
+					.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId())
+					.setHeader(IpHeaders.ACTUAL_CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID))
+					.build());
+		}
+
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -52,7 +52,7 @@ public class TcpNetClientConnectionFactory extends
 	 * @throws Exception
 	 */
 	@Override
-	protected TcpConnection getOrMakeConnection() throws Exception {
+	protected TcpConnection obtainConnection() throws Exception {
 		TcpConnection theConnection = this.getTheConnection();
 		if (theConnection != null && theConnection.isOpen()) {
 			return theConnection;
@@ -70,6 +70,12 @@ public class TcpNetClientConnectionFactory extends
 		return connection;
 	}
 
+	@Override
+	public void start() {
+		this.setActive(true);
+		super.start();
+	}
+
 	/**
 	 * Create a new {@link Socket}. This default implementation uses the default
 	 * {@link SocketFactory}. Override to use some other mechanism
@@ -87,9 +93,6 @@ public class TcpNetClientConnectionFactory extends
 	public void close() {
 	}
 
-	public void run() {
-	}
-
 	protected TcpSocketFactorySupport getTcpSocketFactorySupport() {
 		return tcpSocketFactorySupport;
 	}

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

@@ -19,24 +19,24 @@ package org.springframework.integration.ip.tcp.connection;
 import java.net.Socket;
 import java.net.SocketTimeoutException;
 
-import org.springframework.core.serializer.Serializer;
 import org.springframework.core.serializer.Deserializer;
+import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.Message;
 import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
 
 /**
  * A TcpConnection that uses and underlying {@link Socket}.
- * 
+ *
  * @author Gary Russell
  * @since 2.0
  *
  */
 public class TcpNetConnection extends AbstractTcpConnection {
 
-	private final Socket socket; 
-	
+	private final Socket socket;
+
 	private boolean noReadErrorOnClose;
-	
+
 	/**
 	 * Constructs a TcpNetConnection for the socket.
 	 * @param socket the socket
@@ -47,10 +47,11 @@ public class TcpNetConnection extends AbstractTcpConnection {
 		super(socket, server, lookupHost);
 		this.socket = socket;
 	}
-	
+
 	/**
 	 * Closes this connection.
 	 */
+	@Override
 	public void close() {
 		this.noReadErrorOnClose = true;
 		try {
@@ -77,15 +78,15 @@ public class TcpNetConnection extends AbstractTcpConnection {
 	public int getPort() {
 		return this.socket.getPort();
 	}
-	
+
 	/**
-	 * If there is no listener, and this connection is not for single use, 
+	 * If there is no listener, and this connection is not for single use,
 	 * this method exits. When there is a listener, the method runs in a
 	 * loop reading input from the connections's stream, data is converted
 	 * to an object using the {@link Deserializer} and the listener's
 	 * {@link TcpListener#onMessage(Message)} method is called. For single use
 	 * connections with no listener, the socket is closed after its timeout
-	 * expires. If data is received on a single use socket with no listener, 
+	 * expires. If data is received on a single use socket with no listener,
 	 * a warning is logged.
 	 */
 	public void run() {
@@ -116,7 +117,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
 							else if (logger.isDebugEnabled()) {
 								logger.debug("Read exception " +
 										 this.getConnectionId() + " " +
-										 e.getClass().getSimpleName() + 
+										 e.getClass().getSimpleName() +
 									     ":" + e.getCause() + ":" + e.getMessage());
 							}
 						} else if (logger.isTraceEnabled()) {
@@ -125,7 +126,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
 						} else {
 							logger.error("Read exception " +
 										 this.getConnectionId() + " " +
-										 e.getClass().getSimpleName() + 
+										 e.getClass().getSimpleName() +
 									     ":" + e.getCause() + ":" + e.getMessage());
 						}
 					}
@@ -140,7 +141,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
 					logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
 					continue;
 				}
-				intercepted = listener.onMessage(message);
+				intercepted = this.getListener().onMessage(message);
 			} catch (Exception e) {
 				if (e instanceof NoListenerException) {
 					if (singleUse) {
@@ -151,12 +152,12 @@ public class TcpNetConnection extends AbstractTcpConnection {
 						logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
 					}
 				} else {
-					logger.error("Exception sending meeeage: " + message, e);				
+					logger.error("Exception sending meeeage: " + message, e);
 				}
 			}
 			/*
 			 * For single use sockets, we close after receipt if we are on the client
-			 * side, and the data was not intercepted, 
+			 * side, and the data was not intercepted,
 			 * or the server side has no outbound adapter registered
 			 */
 			if (singleUse && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {

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

@@ -99,13 +99,15 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 				this.harvestClosedConnections();
 			}
 		} catch (Exception e) {
-			this.setListening(false);
 			// don't log an error if we had a good socket once and now it's closed
 			if (e instanceof SocketException && theServerSocket != null) {
 				logger.warn("Server Socket closed");
 			} else if (this.isActive()) {
 				logger.error("Error on ServerSocket", e);
 			}
+		}
+		finally {
+			this.setListening(false);
 			this.setActive(false);
 		}
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -42,18 +42,18 @@ import org.springframework.util.Assert;
  *
  */
 public class TcpNioClientConnectionFactory extends
-		AbstractClientConnectionFactory {
+		AbstractClientConnectionFactory implements Runnable {
 
 	private volatile boolean usingDirectBuffers;
-	
+
 	private volatile Selector selector;
-	
+
 	private final Map<SocketChannel, TcpNioConnection> channelMap = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
-	
+
 	private final BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
 
 	private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport();
-	
+
 	/**
 	 * Creates a TcpNioClientConnectionFactory for connections to the host and port.
 	 * @param host the host
@@ -68,7 +68,8 @@ public class TcpNioClientConnectionFactory extends
 	 * @throws IOException
 	 * @throws SocketException
 	 */
-	protected TcpConnection getOrMakeConnection() throws Exception {
+	@Override
+	protected TcpConnection obtainConnection() throws Exception {
 		int n = 0;
 		while (this.selector == null) {
 			try {
@@ -120,12 +121,24 @@ public class TcpNioClientConnectionFactory extends
 		this.tcpNioConnectionSupport = tcpNioSupport;
 	}
 
+	@Override
 	public void close() {
 		if (this.selector != null) {
 			this.selector.wakeup();
 		}
 	}
 
+	@Override
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.isActive()) {
+				this.setActive(true);
+				this.getTaskExecutor().execute(this);
+			}
+		}
+		super.start();
+	}
+
 	public void run() {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Read selector running for connections to " + this.getHost() + ":" + this.getPort());

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

@@ -98,12 +98,14 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 
 		} catch (IOException e) {
 			this.close();
-			this.setListening(false);
 			if (this.isActive()) {
 				logger.error("Error on ServerSocketChannel", e);
-				this.setActive(false);
 			}
 		}
+		finally {
+			this.setListening(false);
+			this.setActive(false);
+		}
 	}
 
 	/**

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml

@@ -370,4 +370,13 @@
 
 	<int:bridge input-channel="udpAutoChannel" output-channel="nullChannel" />
 
+	<bean id="failCF" class="org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory">
+		<constructor-arg>
+			<list>
+				<ref bean="cfC1"/>
+				<ref bean="cfC2"/>
+			</list>
+		</constructor-arg>
+	</bean>
+
 </beans>

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.ip.tcp;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -164,8 +165,12 @@ public class TcpInboundGatewayTests {
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
 		gateway.start();
-		handler.handleMessage(channel.receive());
-		handler.handleMessage(channel.receive());
+		Message<?> message = channel.receive(10000);
+		assertNotNull(message);
+		handler.handleMessage(message);
+		message = channel.receive(10000);
+		assertNotNull(message);
+		handler.handleMessage(message);
 		latch2.countDown();
 		assertTrue(latch3.await(10, TimeUnit.SECONDS));
 		assertTrue(done.get());
@@ -264,13 +269,13 @@ public class TcpInboundGatewayTests {
 				fail("Failed to listen");
 			}
 		}
-		final SubscribableChannel channel = new DirectChannel(); 
+		final SubscribableChannel channel = new DirectChannel();
 		gateway.setRequestChannel(channel);
 		ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService());
 		channel.subscribe(handler);
 		Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
 		socket1.getOutputStream().write("Test1\r\n".getBytes());
-		Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);		
+		Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);
 		socket2.getOutputStream().write("Test2\r\n".getBytes());
 		byte[] bytes = new byte[errorMessage.length() + 2];
 		readFully(socket1.getInputStream(), bytes);
@@ -299,5 +304,5 @@ public class TcpInboundGatewayTests {
 			buff[i] = (byte) is.read();
 		}
 	}
-	
+
 }

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

@@ -0,0 +1,362 @@
+/*
+ * Copyright 2002-2012 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * 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.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+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;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.SubscribableChannel;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.tcp.TcpInboundGateway;
+import org.springframework.integration.ip.tcp.TcpOutboundGateway;
+import org.springframework.integration.ip.util.SocketTestUtils;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.test.util.TestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class FailoverClientConnectionFactoryTests {
+
+	@Test
+	public void testFailoverGood() throws Exception {
+		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
+		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(factory1);
+		factories.add(factory2);
+		TcpConnection conn1 = makeMockConnection();
+		TcpConnection conn2 = makeMockConnection();
+		when(factory1.getConnection()).thenReturn(conn1);
+		when(factory2.getConnection()).thenReturn(conn2);
+		when(factory1.isActive()).thenReturn(true);
+		when(factory2.isActive()).thenReturn(true);
+		doThrow(new IOException("fail")).when(conn1).send(Mockito.any(Message.class));
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				return null;
+			}
+		}).when(conn2).send(Mockito.any(Message.class));
+		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
+		failoverFactory.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		failoverFactory.getConnection().send(message);
+		Mockito.verify(conn2).send(message);
+	}
+
+	@Test(expected=IOException.class)
+	public void testFailoverAllDead() throws Exception {
+		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
+		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(factory1);
+		factories.add(factory2);
+		TcpConnection conn1 = makeMockConnection();
+		TcpConnection conn2 = makeMockConnection();
+		when(factory1.getConnection()).thenReturn(conn1);
+		when(factory2.getConnection()).thenReturn(conn2);
+		when(factory1.isActive()).thenReturn(true);
+		when(factory2.isActive()).thenReturn(true);
+		doThrow(new IOException("fail")).when(conn1).send(Mockito.any(Message.class));
+		doThrow(new IOException("fail")).when(conn2).send(Mockito.any(Message.class));
+		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
+		failoverFactory.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		failoverFactory.getConnection().send(message);
+		Mockito.verify(conn2).send(message);
+	}
+
+	@Test
+	public void testFailoverAllDeadButOriginalOkAgain() throws Exception {
+		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
+		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(factory1);
+		factories.add(factory2);
+		TcpConnection conn1 = makeMockConnection();
+		TcpConnection conn2 = makeMockConnection();
+		when(factory1.getConnection()).thenReturn(conn1);
+		when(factory2.getConnection()).thenReturn(conn2);
+		when(factory1.isActive()).thenReturn(true);
+		when(factory2.isActive()).thenReturn(true);
+		final AtomicBoolean failedOnce = new AtomicBoolean();
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				if (!failedOnce.get()) {
+					failedOnce.set(true);
+					throw new IOException("fail");
+				}
+				return null;
+			}
+		}).when(conn1).send(Mockito.any(Message.class));
+		doThrow(new IOException("fail")).when(conn2).send(Mockito.any(Message.class));
+		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
+		failoverFactory.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		failoverFactory.getConnection().send(message);
+		Mockito.verify(conn2).send(message);
+		Mockito.verify(conn1, times(2)).send(message);
+	}
+
+	@Test(expected=IOException.class)
+	public void testFailoverConnectNone() throws Exception {
+		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
+		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(factory1);
+		factories.add(factory2);
+		when(factory1.getConnection()).thenThrow(new IOException("fail"));
+		when(factory2.getConnection()).thenThrow(new IOException("fail"));
+		when(factory1.isActive()).thenReturn(true);
+		when(factory2.isActive()).thenReturn(true);
+		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
+		failoverFactory.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		failoverFactory.getConnection().send(message);
+	}
+
+	@Test
+	public void testFailoverConnectToFirstAfterTriedAll() throws Exception {
+		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
+		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(factory1);
+		factories.add(factory2);
+		TcpConnection conn1 = makeMockConnection();
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				return null;
+			}
+		}).when(conn1).send(Mockito.any(Message.class));
+		when(factory1.getConnection()).thenThrow(new IOException("fail")).thenReturn(conn1);
+		when(factory2.getConnection()).thenThrow(new IOException("fail"));
+		when(factory1.isActive()).thenReturn(true);
+		when(factory2.isActive()).thenReturn(true);
+		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
+		failoverFactory.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		failoverFactory.getConnection().send(message);
+		Mockito.verify(conn1).send(message);
+	}
+
+	@Test
+	public void testOkAgainAfterCompleteFailure() throws Exception {
+		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
+		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(factory1);
+		factories.add(factory2);
+		TcpConnection conn1 = makeMockConnection();
+		TcpConnection conn2 = makeMockConnection();
+		when(factory1.getConnection()).thenReturn(conn1);
+		when(factory2.getConnection()).thenReturn(conn2);
+		when(factory1.isActive()).thenReturn(true);
+		when(factory2.isActive()).thenReturn(true);
+		final AtomicInteger failCount = new AtomicInteger();
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				if (failCount.incrementAndGet() < 3) {
+					throw new IOException("fail");
+				}
+				return null;
+			}
+		}).when(conn1).send(Mockito.any(Message.class));
+		doThrow(new IOException("fail")).when(conn2).send(Mockito.any(Message.class));
+		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
+		failoverFactory.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		try {
+			failoverFactory.getConnection().send(message);
+			fail("ExpectedFailure");
+		}
+		catch (IOException e) {}
+		failoverFactory.getConnection().send(message);
+		Mockito.verify(conn2).send(message);
+		Mockito.verify(conn1, times(3)).send(message);
+	}
+
+	public TcpConnection makeMockConnection() {
+		TcpConnection connection = mock(TcpConnection.class);
+		when(connection.isOpen()).thenReturn(true);
+		return connection;
+	}
+
+	@Test
+	public void testRealNet() throws Exception {
+		int port1 = SocketTestUtils.findAvailableServerSocket();
+		int port2 = SocketTestUtils.findAvailableServerSocket(port1 + 1);
+		AbstractClientConnectionFactory client1 = new TcpNetClientConnectionFactory("localhost", port1);
+		AbstractClientConnectionFactory client2 = new TcpNetClientConnectionFactory("localhost", port2);
+		AbstractServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1);
+		AbstractServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2);
+		testRealGuts(client1, client2, server1, server2);
+	}
+
+	@Test
+	public void testRealNio() throws Exception {
+		int port1 = SocketTestUtils.findAvailableServerSocket();
+		int port2 = SocketTestUtils.findAvailableServerSocket(port1 + 1);
+		AbstractClientConnectionFactory client1 = new TcpNioClientConnectionFactory("localhost", port1);
+		AbstractClientConnectionFactory client2 = new TcpNioClientConnectionFactory("localhost", port2);
+		AbstractServerConnectionFactory server1 = new TcpNioServerConnectionFactory(port1);
+		AbstractServerConnectionFactory server2 = new TcpNioServerConnectionFactory(port2);
+		testRealGuts(client1, client2, server1, server2);
+	}
+
+	@Test
+	public void testRealNetSingleUse() throws Exception {
+		int port1 = SocketTestUtils.findAvailableServerSocket();
+		int port2 = SocketTestUtils.findAvailableServerSocket(port1 + 1);
+		AbstractClientConnectionFactory client1 = new TcpNetClientConnectionFactory("localhost", port1);
+		AbstractClientConnectionFactory client2 = new TcpNetClientConnectionFactory("localhost", port2);
+		AbstractServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1);
+		AbstractServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2);
+		client1.setSingleUse(true);
+		client2.setSingleUse(true);
+		testRealGuts(client1, client2, server1, server2);
+	}
+
+	@Test
+	public void testRealNioSingleUse() throws Exception {
+		int port1 = SocketTestUtils.findAvailableServerSocket();
+		int port2 = SocketTestUtils.findAvailableServerSocket(port1 + 1);
+		AbstractClientConnectionFactory client1 = new TcpNioClientConnectionFactory("localhost", port1);
+		AbstractClientConnectionFactory client2 = new TcpNioClientConnectionFactory("localhost", port2);
+		AbstractServerConnectionFactory server1 = new TcpNioServerConnectionFactory(port1);
+		AbstractServerConnectionFactory server2 = new TcpNioServerConnectionFactory(port2);
+		client1.setSingleUse(true);
+		client2.setSingleUse(true);
+		testRealGuts(client1, client2, server1, server2);
+	}
+
+	private void testRealGuts(AbstractClientConnectionFactory client1, AbstractClientConnectionFactory client2,
+			AbstractServerConnectionFactory server1, AbstractServerConnectionFactory server2) throws Exception {
+		int port1;
+		int port2;
+		Executor exec = Executors.newCachedThreadPool();
+		client1.setTaskExecutor(exec);
+		client2.setTaskExecutor(exec);
+		server1.setTaskExecutor(exec);
+		server2.setTaskExecutor(exec);
+		TcpInboundGateway gateway1 = new TcpInboundGateway();
+		gateway1.setConnectionFactory(server1);
+		SubscribableChannel channel = new DirectChannel();
+		final AtomicReference<String> connectionId = new AtomicReference<String>();
+		channel.subscribe(new MessageHandler() {
+			public void handleMessage(Message<?> message) throws MessagingException {
+				connectionId.set((String) message.getHeaders().get(IpHeaders.CONNECTION_ID));
+				((MessageChannel) message.getHeaders().getReplyChannel()).send(message);
+			}
+		});
+		gateway1.setRequestChannel(channel);
+		gateway1.start();
+		TcpInboundGateway gateway2 = new TcpInboundGateway();
+		gateway2.setConnectionFactory(server2);
+		gateway2.setRequestChannel(channel);
+		gateway2.start();
+		waitListening(server1);
+		waitListening(server2);
+		List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
+		factories.add(client1);
+		factories.add(client2);
+		FailoverClientConnectionFactory failFactory = new FailoverClientConnectionFactory(factories);
+		boolean singleUse = client1.isSingleUse();
+		failFactory.setSingleUse(singleUse);
+		failFactory.afterPropertiesSet();
+		TcpOutboundGateway outGateway = new TcpOutboundGateway();
+		outGateway.setConnectionFactory(failFactory);
+		outGateway.start();
+		QueueChannel replyChannel = new QueueChannel();
+		outGateway.setReplyChannel(replyChannel);
+		Message<String> message = new GenericMessage<String>("foo");
+		outGateway.setRemoteTimeout(120000);
+		outGateway.handleMessage(message);
+		Socket socket = getSocket(client1);
+		port1 = socket.getLocalPort();
+		assertTrue(singleUse | connectionId.get().contains(Integer.toString(port1)));
+		Message<?> replyMessage = replyChannel.receive(10000);
+		assertNotNull(replyMessage);
+		server1.stop();
+		waitStopListening(server1);
+		outGateway.handleMessage(message);
+		socket = getSocket(client2);
+		port2 = socket.getLocalPort();
+		assertTrue(singleUse | connectionId.get().contains(Integer.toString(port2)));
+		replyMessage = replyChannel.receive(10000);
+		assertNotNull(replyMessage);
+		gateway2.stop();
+	}
+
+	private Socket getSocket(AbstractClientConnectionFactory client) throws Exception {
+		if (client instanceof TcpNetClientConnectionFactory) {
+			return TestUtils.getPropertyValue(client.getConnection(), "socket", Socket.class);
+		}
+		else {
+			return TestUtils.getPropertyValue(client.getConnection(), "socketChannel", SocketChannel.class).socket();
+		}
+
+	}
+
+	private void waitListening(AbstractServerConnectionFactory scf) throws Exception {
+		int n = 0;
+		while (!scf.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 200) {
+				fail("Failed to listen");
+			}
+		}
+	}
+
+	private void waitStopListening(AbstractServerConnectionFactory scf) throws Exception {
+		int n = 0;
+		while (scf.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 200) {
+				fail("Failed to stop listening");
+			}
+		}
+	}
+}
+