INT-1770 Support Client Mode For TCP Endpoints

Add client mode for inbound endpoints, and for the
outbound adapter.

INT-1770 Allow Inbound Adapter to Open Connection

Normally, inbound adapters use server sockets and wait for incoming
connection. There are use cases where the adapter should establish
the connection and wait for inbound messages.

INT-1770 Open Connection on Start

Allow configuration of outbound adapter to permit
connection establishment when the adapter is started
rather than when the first message arrives.

INT-1770 Allow Inbound Gateway to Open Connection

Normally, inbound gateways listen for connections. There are
use cases where an inbound gateway might open the connection
and then wait for incoming requests.

INT-1770 Parsers

Update parsers and tests to support attributes for setting
endpoints in client-mode.

INT-1770 Docs

Update reference with client-mode information.

INT-1770 Add Control Bus for Client Mode

Enable control bus commands to check status and to attempt
connection establishment.

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -98,21 +98,27 @@ public abstract class IpAdapterParserUtils {
 	static final String TCP_CONNECTION_FACTORY = "connection-factory";
 
 	public static final String INTERCEPTOR_FACTORY_CHAIN = "interceptor-factory-chain";
-	
+
 	public static final String REQUEST_TIMEOUT = "request-timeout";
-	
+
 	public static final String REPLY_TIMEOUT = "reply-timeout";
-	
+
 	public static final String REPLY_CHANNEL = "reply-channel";
 
 	public static final String LOOKUP_HOST = "lookup-host";
-	
+
 	public static final String AUTO_STARTUP = "auto-startup";
 
 	public static final String PHASE = "phase";
 
 	public static final String APPLY_SEQUENCE = "apply-sequence";
 
+	public static final String CLIENT_MODE = "client-mode";
+
+	public static final String RETRY_INTERVAL = "retry-interval";
+
+	public static final String SCHEDULER = "scheduler";
+
 	/**
 	 * Adds a constructor-arg to the provided bean definition builder 
 	 * with the value of the attribute whose name is provided if that

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java

@@ -46,6 +46,12 @@ public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser
 				IpAdapterParserUtils.AUTO_STARTUP);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.PHASE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.CLIENT_MODE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.RETRY_INTERVAL);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SCHEDULER);
 		return builder.getBeanDefinition();
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java

@@ -37,12 +37,16 @@ public class TcpInboundGatewayParser extends AbstractInboundGatewayParser {
 	@Override
 	protected boolean isEligibleAttribute(String attributeName) {
 		return !attributeName.equals(IpAdapterParserUtils.TCP_CONNECTION_FACTORY)
+				&& !attributeName.equals(IpAdapterParserUtils.SCHEDULER)
 				&& super.isEligibleAttribute(attributeName);
 	}
 
 	@Override
 	protected void doPostProcess(BeanDefinitionBuilder builder, Element element) {
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 
-				IpAdapterParserUtils.TCP_CONNECTION_FACTORY);	}
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SCHEDULER);
+	}
 
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java

@@ -40,6 +40,12 @@ public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap
 				IpAdapterParserUtils.AUTO_STARTUP);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.PHASE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.CLIENT_MODE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.RETRY_INTERVAL);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SCHEDULER);
 		return builder.getBeanDefinition();
 	}
 

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

@@ -17,14 +17,22 @@ package org.springframework.integration.ip.tcp;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.ClientModeCapable;
+import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager;
 import org.springframework.integration.ip.tcp.connection.TcpConnection;
 import org.springframework.integration.ip.tcp.connection.TcpListener;
 import org.springframework.integration.ip.tcp.connection.TcpSender;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.Assert;
 
 /**
  * Inbound Gateway using a server connection factory - threading is controlled by the
@@ -38,12 +46,27 @@ import org.springframework.integration.ip.tcp.connection.TcpSender;
  * @since 2.0
  *
  */
-public class TcpInboundGateway extends MessagingGatewaySupport implements TcpListener, TcpSender {
+public class TcpInboundGateway extends MessagingGatewaySupport implements
+		TcpListener, TcpSender, ClientModeCapable {
+
+	private volatile AbstractServerConnectionFactory serverConnectionFactory;
+
+	private volatile AbstractClientConnectionFactory clientConnectionFactory;
 
-	private AbstractServerConnectionFactory connectionFactory;
-	
 	private Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
 
+	private volatile boolean isClientMode;
+
+	private volatile TaskScheduler scheduler;
+
+	private volatile long retryInterval = 60000;
+
+	private volatile ScheduledFuture<?> scheduledFuture;
+
+	private volatile ClientModeConnectionManager clientModeConnectionManager;
+
+	private volatile boolean active;
+
 	public boolean onMessage(Message<?> message) {
 		Message<?> reply = this.sendAndReceiveMessage(message);
 		if (reply == null) {
@@ -73,16 +96,25 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpLis
 	 * @return true if the associated connection factory is listening.
 	 */
 	public boolean isListening() {
-		return connectionFactory.isListening();
-	
+		return this.serverConnectionFactory == null ? false
+				: this.serverConnectionFactory.isListening();
 	}
 
 	/**
-	 * 
+	 * Must be {@link AbstractClientConnectionFactory} or {@link AbstractServerConnectionFactory}.
+	 *
 	 * @param connectionFactory the Connection Factory
 	 */
-	public void setConnectionFactory(AbstractServerConnectionFactory connectionFactory) {
-		this.connectionFactory = connectionFactory;
+	public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
+		Assert.notNull(connectionFactory, "Connection factory must not be null");
+		if (connectionFactory instanceof AbstractServerConnectionFactory) {
+			this.serverConnectionFactory = (AbstractServerConnectionFactory) connectionFactory;
+		} else if (connectionFactory instanceof AbstractClientConnectionFactory) {
+			this.clientConnectionFactory = (AbstractClientConnectionFactory) connectionFactory;
+		} else {
+			throw new IllegalArgumentException("Connection factory must be either an " +
+					"AbstractServerConnectionFactory or an AbstractClientConnectionFactory");
+		}
 		connectionFactory.registerListener(this);
 		connectionFactory.registerSender(this);
 	}
@@ -98,10 +130,117 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpLis
 		return "ip:tcp-inbound-gateway";
 	}
 
-	/**
-	 * @return the connectionFactory
-	 */
-	protected AbstractServerConnectionFactory getConnectionFactory() {
-		return connectionFactory;
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (this.isClientMode) {
+			Assert.notNull(this.clientConnectionFactory,
+					"For client-mode, connection factory must be type='client'");
+			Assert.isTrue(!this.clientConnectionFactory.isSingleUse(),
+					"For client-mode, connection factory must have single-use='false'");
+		}
 	}
+
+	@Override // protected by super#lifecycleLock
+	protected void doStart() {
+		super.doStart();
+		if (!this.active) {
+			this.active = true;
+			if (this.serverConnectionFactory != null) {
+				this.serverConnectionFactory.start();
+			}
+			if (this.clientConnectionFactory != null) {
+				this.clientConnectionFactory.start();
+			}
+			if (this.isClientMode) {
+				ClientModeConnectionManager manager = new ClientModeConnectionManager(
+						this.clientConnectionFactory);
+				this.clientModeConnectionManager = manager;
+				this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval);
+			}
+		}
+	}
+
+	@Override // protected by super#lifecycleLock
+	protected void doStop() {
+		super.doStop();
+		if (this.active) {
+			this.active = false;
+			if (this.scheduledFuture != null) {
+				this.scheduledFuture.cancel(true);
+			}
+			this.clientModeConnectionManager = null;
+			if (this.clientConnectionFactory != null) {
+				this.clientConnectionFactory.stop();
+			}
+			if (this.serverConnectionFactory != null) {
+				this.serverConnectionFactory.stop();
+			}
+		}
+	}
+
+	/**
+	 * @return the isClientMode
+	 */
+	public boolean isClientMode() {
+		return isClientMode;
+	}
+
+	/**
+	 * @param isClientMode
+	 *            the isClientMode to set
+	 */
+	public void setClientMode(boolean isClientMode) {
+		this.isClientMode = isClientMode;
+	}
+
+	/**
+	 * @return the scheduler
+	 */
+	protected TaskScheduler getScheduler() {
+		if (this.scheduler == null) {
+			ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+			scheduler.initialize();
+			this.scheduler = scheduler;
+		}
+		return scheduler;
+	}
+
+	/**
+	 * @param scheduler
+	 *            the scheduler to set
+	 */
+	public void setScheduler(TaskScheduler scheduler) {
+		this.scheduler = scheduler;
+	}
+
+	/**
+	 * @return the retryInterval
+	 */
+	public long getRetryInterval() {
+		return retryInterval;
+	}
+
+	/**
+	 * @param retryInterval
+	 *            the retryInterval to set
+	 */
+	public void setRetryInterval(long retryInterval) {
+		this.retryInterval = retryInterval;
+	}
+
+	public boolean isClientModeConnected() {
+		if (this.isClientMode && this.clientModeConnectionManager != null) {
+			return this.clientModeConnectionManager.isConnected();
+		} else {
+			return false;
+		}
+	}
+
+	public void retryConnection() {
+		if (this.active && this.isClientMode && this.clientModeConnectionManager != null) {
+			this.clientModeConnectionManager.run();
+		}
+	}
+
 }

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

@@ -15,13 +15,20 @@
  */
 package org.springframework.integration.ip.tcp;
 
+import java.util.concurrent.ScheduledFuture;
+
 import org.springframework.integration.Message;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.ClientModeCapable;
+import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager;
 import org.springframework.integration.ip.tcp.connection.ConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpListener;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.Assert;
 
 /**
  * Tcp inbound channel adapter using a TcpConnection to 
@@ -34,34 +41,75 @@ import org.springframework.integration.ip.tcp.connection.TcpListener;
  *
  */
 public class TcpReceivingChannelAdapter 
-	extends MessageProducerSupport implements TcpListener {
+	extends MessageProducerSupport implements TcpListener, ClientModeCapable {
+
+	private AbstractConnectionFactory clientConnectionFactory;
+
+	private AbstractConnectionFactory serverConnectionFactory;
+
+	private volatile boolean isClientMode;
+
+	private volatile TaskScheduler scheduler;
+
+	private volatile long retryInterval = 60000;
+
+	private volatile ScheduledFuture<?> scheduledFuture;
+
+	private volatile ClientModeConnectionManager clientModeConnectionManager;
+
+	private volatile boolean active;
 
-	private ConnectionFactory clientConnectionFactory;
-	
-	private ConnectionFactory serverConnectionFactory;
-	
 	public boolean onMessage(Message<?> message) {
 		sendMessage(message);
 		return false;
 	}
-	
+
 	@Override
-	protected void doStart() {
-		if (this.serverConnectionFactory != null) {
-			this.serverConnectionFactory.start();
-		}
-		if (this.clientConnectionFactory != null) {
-			this.clientConnectionFactory.start();
+	protected void onInit() {
+		super.onInit();
+		if (this.isClientMode) {
+			Assert.notNull(this.clientConnectionFactory,
+					"For client-mode, connection factory must be type='client'");
+			Assert.isTrue(!this.clientConnectionFactory.isSingleUse(),
+					"For client-mode, connection factory must have single-use='false'");
 		}
 	}
 
-	@Override
-	protected void doStop() {
-		if (this.clientConnectionFactory != null) {
-			this.clientConnectionFactory.stop();
+	@Override // protected by super#lifecycleLock
+	protected void doStart() {
+		super.doStart();
+		if (!this.active) {
+			this.active = true;
+			if (this.serverConnectionFactory != null) {
+				this.serverConnectionFactory.start();
+			}
+			if (this.clientConnectionFactory != null) {
+				this.clientConnectionFactory.start();
+			}
+			if (this.isClientMode) {
+				ClientModeConnectionManager manager = new ClientModeConnectionManager(
+						this.clientConnectionFactory);
+				this.clientModeConnectionManager = manager;
+				this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval);
+			}
 		}
-		if (this.serverConnectionFactory != null) {
-			this.serverConnectionFactory.stop();
+	}
+
+	@Override // protected by super#lifecycleLock
+	protected void doStop() {
+		super.doStop();
+		if (this.active) {
+			this.active = false;
+			if (this.scheduledFuture != null) {
+				this.scheduledFuture.cancel(true);
+			}
+			this.clientModeConnectionManager = null;
+			if (this.clientConnectionFactory != null) {
+				this.clientConnectionFactory.stop();
+			}
+			if (this.serverConnectionFactory != null) {
+				this.serverConnectionFactory.stop();
+			}
 		}
 	}
 
@@ -108,4 +156,69 @@ public class TcpReceivingChannelAdapter
 	protected ConnectionFactory getServerConnectionFactory() {
 		return serverConnectionFactory;
 	}
+
+	/**
+	 * @return the isClientMode
+	 */
+	public boolean isClientMode() {
+		return this.isClientMode;
+	}
+
+	/**
+	 * @param isClientMode
+	 *            the isClientMode to set
+	 */
+	public void setClientMode(boolean isClientMode) {
+		this.isClientMode = isClientMode;
+	}
+
+	/**
+	 * @return the scheduler
+	 */
+	protected TaskScheduler getScheduler() {
+		if (this.scheduler == null) {
+			ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+			scheduler.initialize();
+			this.scheduler = scheduler;
+		}
+		return this.scheduler;
+	}
+
+	/**
+	 * @param scheduler
+	 *            the scheduler to set
+	 */
+	public void setScheduler(TaskScheduler scheduler) {
+		this.scheduler = scheduler;
+	}
+
+	/**
+	 * @return the retryInterval
+	 */
+	public long getRetryInterval() {
+		return this.retryInterval;
+	}
+
+	/**
+	 * @param retryInterval
+	 *            the retryInterval to set
+	 */
+	public void setRetryInterval(long retryInterval) {
+		this.retryInterval = retryInterval;
+	}
+
+	public boolean isClientModeConnected() {
+		if (this.isClientMode && this.clientModeConnectionManager != null) {
+			return this.clientModeConnectionManager.isConnected();
+		} else {
+			return false;
+		}
+	}
+
+	public void retryConnection() {
+		if (this.active && this.isClientMode && this.clientModeConnectionManager != null) {
+			this.clientModeConnectionManager.run();
+		}
+	}
+
 }

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

@@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp;
 import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -30,10 +31,15 @@ import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.ClientModeCapable;
+import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager;
 import org.springframework.integration.ip.tcp.connection.ConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpConnection;
 import org.springframework.integration.ip.tcp.connection.TcpSender;
 import org.springframework.integration.mapping.MessageMappingException;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.Assert;
 
 /**
  * Tcp outbound channel adapter using a TcpConnection to
@@ -44,13 +50,14 @@ import org.springframework.integration.mapping.MessageMappingException;
  * @since 2.0
  *
  */
-public class TcpSendingMessageHandler extends AbstractMessageHandler implements TcpSender, SmartLifecycle {
+public class TcpSendingMessageHandler extends AbstractMessageHandler implements
+		TcpSender, SmartLifecycle, ClientModeCapable {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
-	private volatile ConnectionFactory clientConnectionFactory;
+	private volatile AbstractConnectionFactory clientConnectionFactory;
 
-	private volatile ConnectionFactory serverConnectionFactory;
+	private volatile AbstractConnectionFactory serverConnectionFactory;
 
 	private Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
 
@@ -58,6 +65,20 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 
 	private volatile int phase;
 
+	private volatile boolean isClientMode;
+
+	private volatile TaskScheduler scheduler;
+
+	private volatile long retryInterval = 60000;
+
+	private volatile ScheduledFuture<?> scheduledFuture;
+
+	private volatile ClientModeConnectionManager clientModeConnectionManager;
+
+	protected final Object lifecycleMonitor = new Object();
+
+	private volatile boolean active;
+
 	protected TcpConnection getConnection() {
 		TcpConnection connection = null;
 		if (this.clientConnectionFactory == null) {
@@ -167,21 +188,51 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		return "ip:tcp-outbound-channel-adapter";
 	}
 
-	public void start() {
-		if (this.clientConnectionFactory != null) {
-			this.clientConnectionFactory.start();
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (this.isClientMode) {
+			Assert.notNull(this.clientConnectionFactory,
+					"For client-mode, connection factory must be type='client'");
+			Assert.isTrue(!this.clientConnectionFactory.isSingleUse(),
+					"For client-mode, connection factory must have single-use='false'");
 		}
-		if (this.serverConnectionFactory != null) {
-			this.serverConnectionFactory.start();
+	}
+
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.active) {
+				this.active = true;
+				if (this.clientConnectionFactory != null) {
+					this.clientConnectionFactory.start();
+				}
+				if (this.serverConnectionFactory != null) {
+					this.serverConnectionFactory.start();
+				}
+				if (this.isClientMode) {
+					ClientModeConnectionManager manager = new ClientModeConnectionManager(
+							this.clientConnectionFactory);
+					this.clientModeConnectionManager = manager;
+					this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval);
+				}
+			}
 		}
 	}
 
 	public void stop() {
-		if (this.clientConnectionFactory != null) {
-			this.clientConnectionFactory.stop();
-		}
-		if (this.serverConnectionFactory != null) {
-			this.serverConnectionFactory.stop();
+		synchronized (this.lifecycleMonitor) {
+			if (this.active) {
+				this.active = false;
+				if (this.scheduledFuture != null) {
+					this.scheduledFuture.cancel(true);
+				}
+				if (this.clientConnectionFactory != null) {
+					this.clientConnectionFactory.stop();
+				}
+				if (this.serverConnectionFactory != null) {
+					this.serverConnectionFactory.stop();
+				}
+			}
 		}
 	}
 
@@ -200,11 +251,20 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 	}
 
 	public void stop(Runnable callback) {
-		if (this.clientConnectionFactory != null) {
-			this.clientConnectionFactory.stop(callback);
-		}
-		if (this.serverConnectionFactory != null) {
-			this.serverConnectionFactory.stop(callback);
+		synchronized (this.lifecycleMonitor) {
+			if (this.active) {
+				this.active = false;
+				if (this.scheduledFuture != null) {
+					this.scheduledFuture.cancel(true);
+				}
+				this.clientModeConnectionManager = null;
+				if (this.clientConnectionFactory != null) {
+					this.clientConnectionFactory.stop(callback);
+				}
+				if (this.serverConnectionFactory != null) {
+					this.serverConnectionFactory.stop(callback);
+				}
+			}
 		}
 	}
 
@@ -237,4 +297,68 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		return connections;
 	}
 
+	/**
+	 * @return the isClientMode
+	 */
+	public boolean isClientMode() {
+		return this.isClientMode;
+	}
+
+	/**
+	 * @param isClientMode
+	 *            the isClientMode to set
+	 */
+	public void setClientMode(boolean isClientMode) {
+		this.isClientMode = isClientMode;
+	}
+
+	/**
+	 * @return the scheduler
+	 */
+	protected TaskScheduler getScheduler() {
+		if (this.scheduler == null) {
+			ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+			scheduler.initialize();
+			this.scheduler = scheduler;
+		}
+		return this.scheduler;
+	}
+
+	/**
+	 * @param scheduler
+	 *            the scheduler to set
+	 */
+	public void setScheduler(TaskScheduler scheduler) {
+		this.scheduler = scheduler;
+	}
+
+	/**
+	 * @return the retryInterval
+	 */
+	public long getRetryInterval() {
+		return this.retryInterval;
+	}
+
+	/**
+	 * @param retryInterval
+	 *            the retryInterval to set
+	 */
+	public void setRetryInterval(long retryInterval) {
+		this.retryInterval = retryInterval;
+	}
+
+	public boolean isClientModeConnected() {
+		if (this.isClientMode && this.clientModeConnectionManager != null) {
+			return this.clientModeConnectionManager.isConnected();
+		} else {
+			return false;
+		}
+	}
+
+	public void retryConnection() {
+		if (this.active && this.isClientMode && this.clientModeConnectionManager != null) {
+			this.clientModeConnectionManager.run();
+		}
+	}
+
 }

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

@@ -39,6 +39,26 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 		super(host, port);
 	}
 	
+	/**
+	 * Obtains a connection - if {@link #setSingleUse(boolean)} was called with
+	 * true, a new connection is returned; otherwise a single connection is
+	 * reused for all requests while the connection remains open.
+	 */
+	public TcpConnection getConnection() throws Exception {
+		this.checkActive();
+		if (this.isSingleUse()) {
+			return getOrMakeConnection();
+		} else {
+			synchronized(this) {
+				TcpConnection connection = getOrMakeConnection();
+				this.setTheConnection(connection);
+				return connection;
+			}
+		}
+	}
+
+	protected abstract TcpConnection getOrMakeConnection() 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 
@@ -62,6 +82,10 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 				}
 			}
 		}
+		TcpSender sender = this.getSender();
+		if (sender != null) {
+			connection.registerSender(sender);
+		}
 		connection.setMapper(this.getMapper());
 		connection.setDeserializer(this.getDeserializer());
 		connection.setSerializer(this.getSerializer());

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

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2002-2011 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 org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedOperation;
+
+/**
+ * Edpoints implementing this interface are capable
+ * of running in client-mode. For inbound endpoints,
+ * this means that the endpoint establishes the connection
+ * and then receives incoming data.
+ * <p/>
+ * For an outbound adapter, it means that the adapter
+ * will establish the connection rather than waiting
+ * for a message to cause the connection to be
+ * established.
+ *
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+public interface ClientModeCapable {
+
+	/**
+	 * @return true if the endpoint is running in
+	 * client mode.
+	 */
+	@ManagedAttribute
+	boolean isClientMode();
+
+	/**
+	 * @return true if the endpoint is running in
+	 * client mode.
+	 */
+	@ManagedAttribute
+	boolean isClientModeConnected();
+
+	/**
+	 * Immediately attempt to establish the connection.
+	 */
+	@ManagedOperation
+	void retryConnection();
+
+}

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2002-2011 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 org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Intended to be run on a schedule, simply gets the connection
+ * from a client connection factory each time it is run.
+ * If no connection exists (or it has been closed), the
+ * connection factory will create a new one (if possible).
+ *
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+public class ClientModeConnectionManager implements Runnable {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final AbstractConnectionFactory clientConnectionFactory;
+
+	private volatile TcpConnection lastConnection;
+
+	/**
+	 * @param clientConnectionFactory
+	 */
+	public ClientModeConnectionManager(
+			AbstractConnectionFactory clientConnectionFactory) {
+		Assert.notNull(clientConnectionFactory, "Connection factory cannot be null");
+		this.clientConnectionFactory = clientConnectionFactory;
+	}
+
+	public void run() {
+		synchronized (this.clientConnectionFactory) {
+			try {
+				TcpConnection connection = this.clientConnectionFactory.getConnection();
+				if (connection != lastConnection) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Connection " + connection.getConnectionId() + " established");
+					}
+					lastConnection = connection;
+				} else {
+					if (logger.isTraceEnabled()) {
+						logger.trace("Connection " + connection.getConnectionId() + " still OK");
+					}
+				}
+			} catch (Exception e) {
+				logger.error("Could not establish connection using " + this.clientConnectionFactory, e);
+			}
+		}
+	}
+
+	public boolean isConnected() {
+		return this.lastConnection == null ? false : this.lastConnection.isOpen();
+	}
+
+}

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

@@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp.connection;
 
 import java.io.IOException;
 import java.net.Socket;
+import java.net.SocketException;
 
 import javax.net.SocketFactory;
 
@@ -40,12 +41,12 @@ public class TcpNetClientConnectionFactory extends
 	}
 
 	/**
-	 * Obtains a connection - if {@link #setSingleUse(boolean)} was called with
-	 * true, a new connection is returned; otherwise a single connection is
-	 * reused for all requests while the connection remains open.
+	 * @return
+	 * @throws IOException
+	 * @throws SocketException
+	 * @throws Exception
 	 */
-	public TcpConnection getConnection() throws Exception {
-		this.checkActive();
+	protected TcpConnection getOrMakeConnection() throws Exception {
 		TcpConnection theConnection = this.getTheConnection();
 		if (theConnection != null && theConnection.isOpen()) {
 			return theConnection;
@@ -59,9 +60,6 @@ public class TcpNetClientConnectionFactory extends
 		connection = wrapConnection(connection);
 		initializeConnection(connection, socket);
 		this.getTaskExecutor().execute(connection);
-		if (!this.isSingleUse()) {
-			this.setTheConnection(connection);
-		}
 		this.harvestClosedConnections();
 		return connection;
 	}

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

@@ -16,7 +16,9 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
@@ -55,12 +57,12 @@ public class TcpNioClientConnectionFactory extends
 	}
 
 	/**
-	 * Obtains a connection - if {@link #setSingleUse(boolean)} was called with
-	 * true, a new connection is returned; otherwise a single connection is
-	 * reused for all requests while the connection remains open.
+	 * @return
+	 * @throws Exception
+	 * @throws IOException
+	 * @throws SocketException
 	 */
-	public TcpConnection getConnection() throws Exception {
-		this.checkActive();
+	protected TcpConnection getOrMakeConnection() throws Exception {
 		int n = 0;
 		while (this.selector == null) {
 			try {
@@ -72,8 +74,9 @@ public class TcpNioClientConnectionFactory extends
 				throw new Exception("Factory failed to start");
 			}
 		}
-		if (this.getTheConnection() != null && this.getTheConnection().isOpen()) {
-			return this.getTheConnection();
+		TcpConnection theConnection = this.getTheConnection();
+		if (theConnection != null && theConnection.isOpen()) {
+			return theConnection;
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("Opening new socket channel connection to " + this.getHost() + ":" + this.getPort());
@@ -92,9 +95,6 @@ public class TcpNioClientConnectionFactory extends
 		this.connections.put(socketChannel, connection);
 		newChannels.add(socketChannel);
 		selector.wakeup();
-		if (!this.isSingleUse()) {
-			this.setTheConnection(wrappedConnection);
-		}
 		return wrappedConnection;
 	}
 

spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd

@@ -156,6 +156,17 @@ task executors such as a WorkManagerTaskExecutor.
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="client-mode" type="xsd:string" use="optional" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+							 If set to true, causes the adapter to act as a client with respect to
+							 establishing the connection, rather than listening for incoming connections.
+							 Requires a type="client" connection factory, with single-use set to false.
+							 Defaults to true.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attributeGroup ref="clientModeAttributeGroup" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -198,6 +209,17 @@ task executors such as a WorkManagerTaskExecutor.
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="client-mode" type="xsd:string" use="optional" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+							 If set to true, causes the adapter to establish a connection when started,
+							 rather than when the first message is sent.
+							 Requires a type="client" connection factory, with single-use set to false.
+							 Defaults to true.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attributeGroup ref="clientModeAttributeGroup" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -256,6 +278,17 @@ task executors such as a WorkManagerTaskExecutor.
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="client-mode" type="xsd:string" use="optional" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+							 If set to true, causes the gateway to act as a client with respect to
+							 establishing the connection, rather than listening for incoming connections.
+							 Requires a type="client" connection factory, with single-use set to false.
+							 Defaults to true.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attributeGroup ref="clientModeAttributeGroup" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -523,4 +556,29 @@ default is 0. Values can be negative. See SmartLifeCycle.
 		</xsd:attribute>
 	</xsd:complexType>
 
+	<xsd:attributeGroup name="clientModeAttributeGroup">
+		<xsd:attribute name="retry-interval" type="xsd:string" use="optional" default="60000">
+			<xsd:annotation>
+				<xsd:documentation>
+				 When in client mode, specifies the retry interval, in milliseconds, if a connection
+				 cannot be established. Defaults to 60000.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="scheduler" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+				    When in client mode,
+					provide a reference to the TaskScheduler instance to
+					be used for establishing connections. If not provided, the default
+					will use a thread pool of size 1.
+				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.scheduling.TaskScheduler" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
 </xsd:schema>