Merge pull request #444 from garyrussell/INT-2153

* INT-2153:
  INT-2153 SSL Support

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.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.
@@ -113,6 +113,12 @@ public abstract class IpAdapterParserUtils {
 
 	public static final String SCHEDULER = "scheduler";
 
+	public static final String SSL_CONTEXT_SUPPORT = "ssl-context-support";
+
+	public static final String SOCKET_SUPPORT = "socket-support";
+
+	public static final String SOCKET_FACTORY_SUPPORT = "socket-factory-support";
+
 	private IpAdapterParserUtils() {}
 
 	/**

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.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.
@@ -31,12 +31,22 @@ import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionF
 import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSSLSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioConnectionSupport;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioSSLConnectionSupport;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSocketSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpNioConnectionSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+import org.springframework.util.Assert;
 
 /**
  * Instantiates a TcpN(et|io)(Server|Client)ConnectionFactory, depending
  * on type and using-nio attributes.
- * 
+ *
  * @author Gary Russell
  * @since 2.0.5
  */
@@ -90,9 +100,17 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 
 	private volatile boolean applySequence;
 
+	private volatile TcpSSLContextSupport sslContextSupport;
+
+	private volatile TcpSocketSupport socketSupport = new DefaultTcpSocketSupport();
+
+	private volatile TcpNioConnectionSupport nioConnectionSupport;
+
+	private volatile TcpSocketFactorySupport socketFactorySupport;
+
 	@Override
 	public Class<?> getObjectType() {
-		return this.connectionFactory != null ? this.connectionFactory.getClass() 
+		return this.connectionFactory != null ? this.connectionFactory.getClass()
 											  : AbstractConnectionFactory.class;
 	}
 
@@ -104,12 +122,14 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 				this.setCommonAttributes(connectionFactory);
 				this.setServerAttributes(connectionFactory);
 				connectionFactory.setUsingDirectBuffers(this.usingDirectBuffers);
+				connectionFactory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
 				this.connectionFactory = connectionFactory;
 			} else {
 				TcpNioClientConnectionFactory connectionFactory = new TcpNioClientConnectionFactory(
 						this.host, this.port);
 				this.setCommonAttributes(connectionFactory);
 				connectionFactory.setUsingDirectBuffers(this.usingDirectBuffers);
+				connectionFactory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
 				this.connectionFactory = connectionFactory;
 			}
 		} else {
@@ -117,11 +137,13 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 				TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(this.port);
 				this.setCommonAttributes(connectionFactory);
 				this.setServerAttributes(connectionFactory);
+				connectionFactory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
 				this.connectionFactory = connectionFactory;
 			} else {
 				TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory(
 						this.host, this.port);
 				this.setCommonAttributes(connectionFactory);
+				connectionFactory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
 				this.connectionFactory = connectionFactory;
 			}
 		}
@@ -146,12 +168,38 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 		factory.setSoTrafficClass(this.soTrafficClass);
 		factory.setTaskExecutor(this.taskExecutor);
 		factory.setBeanName(this.beanName);
+		factory.setTcpSocketSupport(this.socketSupport);
 	}
 
 	private void setServerAttributes(AbstractServerConnectionFactory factory) {
 		factory.setLocalAddress(this.localAddress);
 	}
 
+	private TcpSocketFactorySupport obtainSocketFactorySupport() {
+		if (this.socketFactorySupport != null) {
+			return this.socketFactorySupport;
+		}
+		if (this.sslContextSupport == null) {
+			return new DefaultTcpNetSocketFactorySupport();
+		}
+		else {
+			return new DefaultTcpNetSSLSocketFactorySupport(this.sslContextSupport);
+		}
+	}
+
+	private TcpNioConnectionSupport obtainNioConnectionSupport() {
+		if (this.nioConnectionSupport != null) {
+			return this.nioConnectionSupport;
+		}
+		if (this.sslContextSupport == null) {
+			return new DefaultTcpNioConnectionSupport();
+		}
+		else {
+			return new DefaultTcpNioSSLConnectionSupport(this.sslContextSupport);
+		}
+
+	}
+
 	/**
 	 * @param port the port to set
 	 */
@@ -163,6 +211,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 * @param host the host to set
 	 */
 	public void setHost(String host) {
+		Assert.notNull(host, "Host may not be null");
 		this.host = host;
 	}
 
@@ -178,6 +227,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory#setLocalAddress(java.lang.String)
 	 */
 	public void setLocalAddress(String localAddress) {
+		Assert.notNull(localAddress, "LocalAddress may not be null");
 		this.localAddress = localAddress;
 	}
 
@@ -257,6 +307,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setTaskExecutor(java.util.concurrent.Executor)
 	 */
 	public void setTaskExecutor(Executor taskExecutor) {
+		Assert.notNull(taskExecutor, "Executor may not be null");
 		this.taskExecutor = taskExecutor;
 	}
 
@@ -265,6 +316,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setDeserializer(org.springframework.core.serializer.Deserializer)
 	 */
 	public void setDeserializer(Deserializer<?> deserializer) {
+		Assert.notNull(deserializer, "Deserializer may not be null");
 		this.deserializer = deserializer;
 	}
 
@@ -273,6 +325,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSerializer(org.springframework.core.serializer.Serializer)
 	 */
 	public void setSerializer(Serializer<?> serializer) {
+		Assert.notNull(serializer, "Serializer may not be null");
 		this.serializer = serializer;
 	}
 
@@ -281,7 +334,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setMapper(org.springframework.integration.ip.tcp.connection.TcpMessageMapper)
 	 */
 	public void setMapper(TcpMessageMapper mapper) {
-		this.mapper = mapper; 
+		Assert.notNull(mapper, "TcpMessageMapper may not be null");
+		this.mapper = mapper;
 	}
 
 	/**
@@ -306,6 +360,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	 */
 	public void setInterceptorFactoryChain(
 			TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
+		Assert.notNull(interceptorFactoryChain, "InterceptorFactoryChain may not be null");
 		this.interceptorFactoryChain = interceptorFactoryChain;
 	}
 
@@ -318,7 +373,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	}
 
 	/**
-	 * 
+	 *
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#start()
 	 */
 	public void start() {
@@ -326,7 +381,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	}
 
 	/**
-	 * 
+	 *
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#stop()
 	 */
 	public void stop() {
@@ -372,5 +427,30 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 		this.applySequence = applySequence;
 	}
 
+	public void setSslContextSupport(TcpSSLContextSupport sslContextSupport) {
+		Assert.notNull(sslContextSupport, "TcpSSLConstextSupport may not be null");
+		this.sslContextSupport = sslContextSupport;
+	}
+
+	public void setSocketSupport(TcpSocketSupport tcpSocketSupport) {
+		Assert.notNull(tcpSocketSupport, "TcpSocketSupport may not be null");
+		this.socketSupport = tcpSocketSupport;
+	}
+
+	/**
+	 * Rare property - not exposed through namespace
+	 * @param tcpNioSupport
+	 */
+	public void setNioConnectionSupport(TcpNioConnectionSupport tcpNioSupport) {
+		Assert.notNull(tcpNioSupport, "TcpNioConnectionSupport may not be null");
+		this.nioConnectionSupport = tcpNioSupport;
+	}
+
+	public void setSocketFactorySupport(
+			TcpSocketFactorySupport tcpSocketFactorySupport) {
+		Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null");
+		this.socketFactorySupport = tcpSocketFactorySupport;
+	}
+
 }
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.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.
@@ -31,8 +31,6 @@ import org.w3c.dom.Element;
  */
 public class TcpConnectionParser extends AbstractBeanDefinitionParser {
 	
-	private static final String BASE_PACKAGE = "org.springframework.integration.ip.config";
-
 	@Override
 	protected AbstractBeanDefinition parseInternal(Element element,
 			ParserContext parserContext) {
@@ -45,8 +43,7 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
 			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE + 
 					" must be 'client' or 'server' for a TCP Connection Factory", element);
 		}		
-		builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + 
-				".TcpConnectionFactoryFactoryBean");
+		builder = BeanDefinitionBuilder.genericBeanDefinition(TcpConnectionFactoryFactoryBean.class);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "type");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
 				IpAdapterParserUtils.HOST);
@@ -83,7 +80,13 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
 				IpAdapterParserUtils.LOOKUP_HOST);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.APPLY_SEQUENCE);
-	
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SSL_CONTEXT_SUPPORT);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SOCKET_FACTORY_SUPPORT);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SOCKET_SUPPORT);
+
 		return builder.getBeanDefinition();
 	}
 

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

@@ -41,6 +41,8 @@ import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.core.OrderlyShutdownCapable;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSocketSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
 import org.springframework.util.Assert;
 
@@ -100,6 +102,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 
 	private volatile List<TcpConnection> connections = new LinkedList<TcpConnection>();
 
+	private volatile TcpSocketSupport tcpSocketSupport = new DefaultTcpSocketSupport();
+
 	protected final Object lifecycleMonitor = new Object();
 
 	private volatile long nextCheckForClosedNioConnections;
@@ -141,6 +145,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 			socket.setTrafficClass(this.soTrafficClass);
 		}
 		socket.setKeepAlive(this.soKeepAlive);
+		this.tcpSocketSupport.postProcessSocket(socket);
 	}
 
 	/**
@@ -678,4 +683,13 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 		}
 	}
 
+	protected TcpSocketSupport getTcpSocketSupport() {
+		return tcpSocketSupport;
+	}
+
+	public void setTcpSocketSupport(TcpSocketSupport tcpSocketSupport) {
+		Assert.notNull(tcpSocketSupport, "TcpSocketSupport must not be null");
+		this.tcpSocketSupport = tcpSocketSupport;
+	}
+
 }

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

@@ -16,21 +16,22 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
 
-/** 
+/**
  * Base class for all server connection factories. Server connection factories
  * listen on a port for incoming connections and create new TcpConnection objects
  * for each new connection.
- *  
+ *
  * @author Gary Russell
  * @since 2.0
  */
 public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
 
 	private boolean listening;
-	
+
 	private String localAddress;
 
 
@@ -60,18 +61,18 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 
 
 	/**
-	 * 
+	 *
 	 * @return true if the server is listening on the port.
 	 */
 	public boolean isListening() {
 		return listening;
 	}
-	
+
 	/**
 	 * Transfers attributes such as (de)serializer, singleUse etc to a new connection.
 	 * For single use sockets, enforces a socket timeout (default 10 seconds).
 	 * @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();
@@ -98,9 +99,13 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 		}
 
 	}
-	
+
+	protected void postProcessServerSocket(ServerSocket serverSocket) {
+		this.getTcpSocketSupport().postProcessServerSocket(serverSocket);
+	}
+
 	/**
-	 * 
+	 *
 	 * @return the localAddress
 	 */
 	public String getLocalAddress() {
@@ -108,7 +113,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 	}
 
 	/**
-	 * Used on multi-homed systems to enforce the server to listen 
+	 * Used on multi-homed systems to enforce the server to listen
 	 * on a specfic network address instead of all network adapters.
 	 * @param localAddress the ip address of the required adapter.
 	 */

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

@@ -22,8 +22,12 @@ import java.net.SocketException;
 
 import javax.net.SocketFactory;
 
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport;
+import org.springframework.util.Assert;
+
 /**
- * A client connection factory that creates {@link TcpNetConnection}s. 
+ * A client connection factory that creates {@link TcpNetConnection}s.
  * @author Gary Russell
  * @since 2.0
  *
@@ -31,6 +35,8 @@ import javax.net.SocketFactory;
 public class TcpNetClientConnectionFactory extends
 		AbstractClientConnectionFactory {
 
+	private volatile TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport();
+
 	/**
 	 * Creates a TcpNetClientConnectionFactory for connections to the host and port.
 	 * @param host the host
@@ -45,6 +51,7 @@ public class TcpNetClientConnectionFactory extends
 	 * @throws SocketException
 	 * @throws Exception
 	 */
+	@Override
 	protected TcpConnection getOrMakeConnection() throws Exception {
 		TcpConnection theConnection = this.getTheConnection();
 		if (theConnection != null && theConnection.isOpen()) {
@@ -73,13 +80,24 @@ public class TcpNetClientConnectionFactory extends
 	 * @throws IOException
 	 */
 	protected Socket createSocket(String host, int port) throws IOException {
-		return SocketFactory.getDefault().createSocket(host, port);
+		return this.tcpSocketFactorySupport.getSocketFactory().createSocket(host, port);
 	}
 
+	@Override
 	public void close() {
 	}
 
 	public void run() {
 	}
 
+	protected TcpSocketFactorySupport getTcpSocketFactorySupport() {
+		return tcpSocketFactorySupport;
+	}
+
+	public void setTcpSocketFactorySupport(
+			TcpSocketFactorySupport tcpSocketFactorySupport) {
+		Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null");
+		this.tcpSocketFactorySupport = tcpSocketFactorySupport;
+	}
+
 }

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

@@ -109,7 +109,11 @@ public class TcpNetConnection extends AbstractTcpConnection {
 						logger.debug("Closing single use socket after timeout");
 					} else {
 						if (this.noReadErrorOnClose) {
-							if (logger.isDebugEnabled()) {
+							if (logger.isTraceEnabled()) {
+								logger.trace("Read exception " +
+										 this.getConnectionId(), e);
+							}
+							else if (logger.isDebugEnabled()) {
 								logger.debug("Read exception " +
 										 this.getConnectionId() + " " +
 										 e.getClass().getSimpleName() + 

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

@@ -21,9 +21,14 @@ import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
+import java.net.SocketTimeoutException;
 
 import javax.net.ServerSocketFactory;
 
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport;
+import org.springframework.util.Assert;
+
 /**
  * Implements a server connection factory that produces {@link TcpNetConnection}s using
  * a {@link ServerSocket}. Must have a {@link TcpListener} registered.
@@ -33,7 +38,9 @@ import javax.net.ServerSocketFactory;
  */
 public class TcpNetServerConnectionFactory extends AbstractServerConnectionFactory {
 
-	private ServerSocket serverSocket;
+	private volatile ServerSocket serverSocket;
+
+	private volatile TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport();
 
 	/**
 	 * Listens for incoming connections on the port.
@@ -63,12 +70,27 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
 				theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), whichNic);
 			}
+			this.getTcpSocketSupport().postProcessServerSocket(theServerSocket);
 			this.serverSocket = theServerSocket;
 			this.setListening(true);
 			logger.info("Listening on port " + this.getPort());
 			while (true) {
-				final Socket socket = serverSocket.accept();
-				logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
+				final Socket socket;
+				/*
+				 *  User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT.
+				 *  Not fatal.
+				 */
+				try {
+					socket = serverSocket.accept();
+				} catch (SocketTimeoutException ste) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Timed out on accept; continuing");
+					}
+					continue;
+				}
+				if (logger.isDebugEnabled()) {
+					logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
+				}
 				setSocketAttributes(socket);
 				TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost());
 				connection = wrapConnection(connection);
@@ -100,11 +122,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 	 * @throws IOException
 	 */
 	protected ServerSocket createServerSocket(int port, int backlog, InetAddress whichNic) throws IOException {
+		ServerSocketFactory serverSocketFactory = this.tcpSocketFactorySupport.getServerSocketFactory();
 		if (whichNic == null) {
-			return ServerSocketFactory.getDefault().createServerSocket(port,
+			return serverSocketFactory.createServerSocket(port,
 					Math.abs(backlog));
 		} else {
-			return ServerSocketFactory.getDefault().createServerSocket(port,
+			return serverSocketFactory.createServerSocket(port,
 					Math.abs(backlog), whichNic);
 		}
 	}
@@ -125,5 +148,15 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 	protected ServerSocket getServerSocket() {
 		return serverSocket;
 	}
+
+	protected TcpSocketFactorySupport getTcpSocketFactorySupport() {
+		return tcpSocketFactorySupport;
+	}
+
+	public void setTcpSocketFactorySupport(
+			TcpSocketFactorySupport tcpSocketFactorySupport) {
+		Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null");
+		this.tcpSocketFactorySupport = tcpSocketFactorySupport;
+	}
 	
 }

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

@@ -30,6 +30,10 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioConnectionSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpNioConnectionSupport;
+import org.springframework.util.Assert;
+
 
 /**
  * A client connection factory that creates {@link TcpNioConnection}s.
@@ -40,14 +44,15 @@ import java.util.concurrent.LinkedBlockingQueue;
 public class TcpNioClientConnectionFactory extends
 		AbstractClientConnectionFactory {
 
-	private boolean usingDirectBuffers;
+	private volatile boolean usingDirectBuffers;
 	
-	private Selector selector;
+	private volatile Selector selector;
 	
-	private Map<SocketChannel, TcpNioConnection> channelMap = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
+	private final Map<SocketChannel, TcpNioConnection> channelMap = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
 	
-	private BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
+	private final BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
 
+	private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport();
 	
 	/**
 	 * Creates a TcpNioClientConnectionFactory for connections to the host and port.
@@ -84,7 +89,8 @@ public class TcpNioClientConnectionFactory extends
 		}
 		SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort()));
 		setSocketAttributes(socketChannel.socket());
-		TcpNioConnection connection = new TcpNioConnection(socketChannel, false, this.isLookupHost());
+		TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(
+				socketChannel, false, this.isLookupHost());
 		connection.setUsingDirectBuffers(this.usingDirectBuffers);
 		connection.setTaskExecutor(this.getTaskExecutor());
 		TcpConnection wrappedConnection = wrapConnection(connection);
@@ -109,6 +115,11 @@ public class TcpNioClientConnectionFactory extends
 		this.usingDirectBuffers = usingDirectBuffers;
 	}
 
+	public void setTcpNioConnectionSupport(TcpNioConnectionSupport tcpNioSupport) {
+		Assert.notNull(tcpNioSupport, "TcpNioSupport must not be null");
+		this.tcpNioConnectionSupport = tcpNioSupport;
+	}
+
 	public void close() {
 		if (this.selector != null) {
 			this.selector.wakeup();

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

@@ -33,10 +33,11 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.Message;
 import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
+import org.springframework.util.Assert;
 
 /**
  * A TcpConnection that uses and underlying {@link SocketChannel}.
- * 
+ *
  * @author Gary Russell
  * @since 2.0
  *
@@ -45,7 +46,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private final SocketChannel socketChannel;
 
-	private volatile OutputStream channelOutputStream;
+	private final ChannelOutputStream channelOutputStream;
 
 	private volatile PipedOutputStream pipedOutputStream;
 
@@ -79,6 +80,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		this.channelOutputStream = new ChannelOutputStream();
 	}
 
+	@Override
 	public void close() {
 		doClose();
 	}
@@ -103,7 +105,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	public void send(Message<?> message) throws Exception {
 		synchronized(this.getMapper()) {
 			Object object = this.getMapper().fromMessage(message);
-			((Serializer<Object>) this.getSerializer()).serialize(object, this.channelOutputStream);
+			((Serializer<Object>) this.getSerializer()).serialize(object, this.getChannelOutputStream());
 			this.afterSend(message);
 		}
 	}
@@ -131,9 +133,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	/**
-	 * 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, this method assembles
-	 * data into messages by invoking convertAndSend whenever there is 
+	 * data into messages by invoking convertAndSend whenever there is
 	 * data in the input Stream. Method exits when a message is complete
 	 * and there is no more data; thus freeing the thread to work on other
 	 * sockets.
@@ -170,7 +172,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 				} else {
 					logger.error("Read exception " +
 								 this.getConnectionId() + " " +
-								 e.getClass().getSimpleName() + 
+								 e.getClass().getSimpleName() +
 							     ":" + e.getCause() + ":" + e.getMessage());
 				}
 				this.closeConnection();
@@ -244,7 +246,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		}
 		/*
 		 * 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 (this.isSingleUse() && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {
@@ -267,23 +269,37 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			}
 			// If there is no assembler running, start one
 			checkForAssembler();
-			this.rawBuffer.clear();
+			if (logger.isTraceEnabled()) {
+				logger.trace("Before read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
+			}
 			int len = this.socketChannel.read(this.rawBuffer);
 			if (len < 0) {
 				this.writingToPipe = false;
 				this.closeConnection();
 			}
+			if (logger.isTraceEnabled()) {
+				logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
+			}
 			this.rawBuffer.flip();
+			if (logger.isTraceEnabled()) {
+				logger.trace("After flip:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
+			}
 			if (logger.isDebugEnabled()) {
 				logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
 			}
-			this.pipedOutputStream.write(this.rawBuffer.array(), 0, this.rawBuffer.limit());
-			this.pipedOutputStream.flush();
+			this.sendToPipe(this.rawBuffer);
 		} finally {
 			this.writingToPipe = false;
 		}
 	}
 
+	protected void sendToPipe(ByteBuffer rawBuffer) throws IOException {
+		Assert.notNull(rawBuffer, "rawBuffer cannot be null");
+		this.pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
+		this.pipedOutputStream.flush();
+		rawBuffer.clear();
+	}
+
 	private void checkForAssembler() {
 		synchronized(this.executionControl) {
 			if (this.executionControl.incrementAndGet() <= 1) {
@@ -311,9 +327,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			}
 			this.closeConnection();
 		} catch (Exception e) {
-			logger.error("Exception on Read " + 
-					     this.getConnectionId() + " " + 
-					     e.getMessage());
+			logger.error("Exception on Read " +
+					     this.getConnectionId() + " " +
+					     e.getMessage(), e);
 			this.closeConnection();
 		}
 	}
@@ -326,7 +342,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	/**
-	 * 
+	 *
 	 * @param taskExecutor the taskExecutor to set
 	 */
 	public void setTaskExecutor(Executor taskExecutor) {
@@ -342,8 +358,16 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		this.usingDirectBuffers = usingDirectBuffers;
 	}
 
+	protected boolean isUsingDirectBuffers() {
+		return usingDirectBuffers;
+	}
+
+	protected ChannelOutputStream getChannelOutputStream() {
+		return channelOutputStream;
+	}
+
 	/**
-	 * 
+	 *
 	 * @return Time of last read.
 	 */
 	public long getLastRead() {
@@ -351,7 +375,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	/**
-	 * 
+	 *
 	 * @param lastRead The time of the last read.
 	 */
 	public void setLastRead(long lastRead) {
@@ -359,7 +383,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	/**
-	 * OutputStream to wrap a SocketChannel; implements timeout on write. 
+	 * OutputStream to wrap a SocketChannel; implements timeout on write.
 	 *
 	 */
 	class ChannelOutputStream extends OutputStream {
@@ -397,7 +421,10 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			doWrite(buffer);
 		}
 
-		private synchronized void doWrite(ByteBuffer buffer) throws IOException {
+		protected synchronized void doWrite(ByteBuffer buffer) throws IOException {
+			if (logger.isDebugEnabled()) {
+				logger.debug(getConnectionId() + " writing " + buffer.remaining());
+			}
 			socketChannel.write(buffer);
 			int remaining = buffer.remaining();
 			if (remaining == 0) {

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

@@ -0,0 +1,396 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import javax.net.ssl.SSLEngineResult.Status;
+import javax.net.ssl.SSLException;
+
+import org.springframework.integration.MessagingException;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link TcpConnection} supporting SSL/TLS over NIO.
+ * Unlike TcpNetConnection, which uses Sockets, the JVM does not directly support SSL for
+ * SocketChannels, used by NIO. Instead, the SSLEngine is provided whereby the SSL
+ * encryption is performed by passing in a plain text buffer, and receiving an
+ * encrypted buffer to transmit over the network. Similarly, encrypted data read from
+ * the network is decrypted.<p>
+ * However, before this can be done, certain handshaking operations are required, involving
+ * the creation of data buffers which must be exchanged by the peers. A number of such
+ * transfers are required; once the handshake is finished, it is relatively simple to
+ * encrypt/decrypt the data.<p>
+ * Also, it may be deemed necessary to re-perform handshaking.<p>
+ * This class supports the management of handshaking as necessary, both from the
+ * initiating and receiving peers.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class TcpNioSSLConnection extends TcpNioConnection {
+
+	private final SSLEngine sslEngine;
+
+	private volatile ByteBuffer decoded;
+
+	private volatile ByteBuffer encoded;
+
+	private volatile SSLChannelOutputStream sslChannelOutputStream;
+
+	private final Semaphore semaphore = new Semaphore(0);
+
+	private final Object monitorLock = new Object();
+
+	private volatile boolean writerActive;
+
+	private boolean needMoreNetworkData;
+
+	public TcpNioSSLConnection(SocketChannel socketChannel, boolean server,
+			boolean lookupHost, SSLEngine sslEngine) throws Exception {
+		super(socketChannel, server, lookupHost);
+		this.sslEngine = sslEngine;
+	}
+
+	/**
+	 * Overrides super class method to perform decryption and/or participate
+	 * in handshaking. Decrypted data is sent to the super class to be
+	 * assembled into a Message. Data received from the network may
+	 * constitute multiple SSL packets, and may end with a partial
+	 * packet. In that case, the buffer is compacted, ready to receive
+	 * the remainder of the packet.
+	 */
+	@Override
+	protected void sendToPipe(final ByteBuffer networkBuffer) throws IOException {
+		Assert.notNull(networkBuffer, "rawBuffer cannot be null");
+		if (logger.isDebugEnabled()) {
+			logger.debug("sendToPipe " + sslEngine.getHandshakeStatus() + ", remaining:" + networkBuffer.remaining());
+		}
+		SSLEngineResult result = null;
+		while (!this.needMoreNetworkData) {
+			result = decode(networkBuffer);
+			if (logger.isDebugEnabled()) {
+				logger.debug("result " + resultToString(result) + ", remaining:" + networkBuffer.remaining());
+			}
+		}
+		this.needMoreNetworkData = false;
+		if (result.getStatus() == Status.BUFFER_UNDERFLOW) {
+			networkBuffer.compact();
+		}
+		else {
+			networkBuffer.clear();
+		}
+		if (logger.isDebugEnabled()) {
+			logger.debug("sendToPipe.x " + resultToString(result) + ", remaining:" + networkBuffer.remaining());
+		}
+	}
+
+	/**
+	 * Performs the actual decryption of a received packet - which may be real
+	 * data, or handshaking data. Appropriate action is taken with the data.
+	 * If this side did not initiate the handshake, any handshaking data sent out
+	 * is handled by the thread running in the {@link SSLChannelOutputStream#doWrite(ByteBuffer)}
+	 * method, which is awoken here, as a result of reaching that stage in the handshaking.
+	 */
+	private SSLEngineResult decode(ByteBuffer networkBuffer) throws IOException {
+		SSLEngineResult result = new SSLEngineResult(Status.OK, this.sslEngine.getHandshakeStatus(), 0, 0);
+		HandshakeStatus handshakeStatus = this.sslEngine.getHandshakeStatus();
+		switch (handshakeStatus) {
+		case NEED_TASK:
+			runTasks();
+			break;
+		case NEED_UNWRAP:
+		case FINISHED:
+		case NOT_HANDSHAKING:
+			this.decoded.clear();
+			result = this.sslEngine.unwrap(networkBuffer, this.decoded);
+			if (logger.isDebugEnabled()) {
+				logger.debug("After unwrap:" + resultToString(result));
+			}
+			Status status = result.getStatus();
+			if (status == Status.BUFFER_OVERFLOW) {
+				this.decoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize());
+			}
+			if (result.bytesProduced() > 0) {
+				this.decoded.flip();
+				super.sendToPipe(this.decoded);
+			}
+			break;
+		case NEED_WRAP:
+			if (!resumeWriterIfNeeded()) {
+				this.encoded.clear();
+				result = this.sslEngine.wrap(networkBuffer, this.encoded);
+				if (logger.isDebugEnabled()) {
+					logger.debug("After wrap:" + resultToString(result));
+				}
+				if (result.getStatus() == Status.BUFFER_OVERFLOW) {
+					this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize());
+				}
+				else {
+					this.encoded.flip();
+					getSSLChannelOutputStream().writeEncoded(this.encoded);
+				}
+			}
+			break;
+		default:
+		}
+		switch (result.getHandshakeStatus()) {
+		case FINISHED:
+			resumeWriterIfNeeded();
+			// switch fall-through intended
+		case NOT_HANDSHAKING:
+		case NEED_UNWRAP:
+			this.needMoreNetworkData = result.getStatus() == Status.BUFFER_UNDERFLOW || networkBuffer.remaining() == 0;
+			break;
+		default:
+		}
+		return result;
+	}
+
+	/**
+	 * Handshake sends are handled by the initiator.
+	 * @return false if we are the initiator.
+	 */
+	private boolean resumeWriterIfNeeded() {
+		if (this.writerActive) {
+			if (logger.isTraceEnabled()) {
+				logger.trace("Waking sender, permits:" + this.semaphore.availablePermits());
+			}
+			this.semaphore.release();
+			return true;
+		}
+		return false;
+	}
+
+	/**
+	 * Part of the SSLEngine handshaking protocol required at
+	 * various stages. Tasks are run on the current thread.
+	 */
+	private void runTasks() {
+		Runnable task;
+		while ((task = this.sslEngine.getDelegatedTask()) != null) {
+			task.run();
+		}
+	}
+
+	/**
+	 * Determines whether {@link #runTasks()} is needed and invokes if so.
+	 */
+	private HandshakeStatus runTasksIfNeeded(SSLEngineResult result) throws IOException {
+		if (result != null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Running tasks if needed " + resultToString(result));
+			}
+			if (result.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
+				runTasks();
+			}
+		}
+		HandshakeStatus handshakeStatus = sslEngine.getHandshakeStatus();
+		if (logger.isDebugEnabled()) {
+			logger.debug("New handshake status " + handshakeStatus);
+		}
+		return handshakeStatus;
+	}
+
+	/**
+	 * Initializes the SSLEngine and sets up the encryption/decryption buffers.
+	 */
+	public void init() throws IOException {
+		if (this.decoded == null) {
+			this.decoded = allocateEncryptionBuffer(2048);
+			this.encoded = allocateEncryptionBuffer(2048);
+			this.initilizeEngine();
+		}
+	}
+
+	private ByteBuffer allocateEncryptionBuffer(int size) {
+		if (this.isUsingDirectBuffers()) {
+			return ByteBuffer.allocateDirect(size);
+		}
+		else {
+			return ByteBuffer.allocate(size);
+		}
+	}
+
+	private void initilizeEngine() throws IOException {
+		boolean client = !this.isServer();
+		this.sslEngine.setUseClientMode(client);
+	}
+
+	@Override
+	protected ChannelOutputStream getChannelOutputStream() {
+		synchronized (this.monitorLock) {
+			if (this.sslChannelOutputStream == null) {
+				this.sslChannelOutputStream = new SSLChannelOutputStream(super.getChannelOutputStream());
+			}
+			return this.sslChannelOutputStream;
+		}
+	}
+
+	protected SSLChannelOutputStream getSSLChannelOutputStream() {
+		if (this.sslChannelOutputStream == null) {
+			return (SSLChannelOutputStream) this.getChannelOutputStream();
+		}
+		else {
+			return this.sslChannelOutputStream;
+		}
+	}
+
+	private String resultToString(SSLEngineResult result) {
+		return result.toString().replace('\n', ' ');
+	}
+
+	/**
+	 * Subclass of {@link TcpNioConnection.ChannelOutputStream} to handle encryption
+	 * of outbound data. Wraps an instance of the superclass, which is invoked to
+	 * send to encrypted data to the SocketChannel.
+	 *
+	 */
+	class SSLChannelOutputStream extends ChannelOutputStream  {
+
+		private final ChannelOutputStream channelOutputStream;
+
+		public SSLChannelOutputStream(ChannelOutputStream channelOutputStream) {
+			this.channelOutputStream = channelOutputStream;
+		}
+
+		/**
+		 * Encrypts the plaintText buffer and writes it to the SocketChannel.
+		 * Will participate in SSL handshaking as necessary. For very large
+		 * data, the SSL packets will be limited by the engine's buffer sizes
+		 * and multiple writes will be necessary.
+		 */
+		@Override
+		protected synchronized void doWrite(ByteBuffer plainText)
+				throws IOException {
+			try {
+				TcpNioSSLConnection.this.writerActive = true;
+				int remaining = plainText.remaining();
+				while (remaining > 0) {
+					SSLEngineResult result = encode(plainText);
+					if (logger.isDebugEnabled()) {
+						logger.debug("doWrite: " + resultToString(result));
+					}
+					if (result.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING) {
+						writeEncodedIfAny();
+						if (plainText.remaining() >= remaining) {
+							throw new MessagingException(
+									"Unexpected condition - SSL wrap did not consume any data; remaining = "
+											+ remaining);
+						}
+						remaining = plainText.remaining();
+					} else {
+						doClientSideHandshake(plainText, result);
+						writeEncodedIfAny();
+					}
+				}
+			}
+			finally {
+				TcpNioSSLConnection.this.writerActive = false;
+			}
+		}
+
+		/**
+		 * Handles SSL handshaking; when network data is needed from the peer, suspends
+		 * until that data is received.
+		 */
+		private void doClientSideHandshake(ByteBuffer plainText,
+				SSLEngineResult result) throws IOException, SSLException {
+			TcpNioSSLConnection.this.semaphore.drainPermits();
+			HandshakeStatus status = TcpNioSSLConnection.this.sslEngine.getHandshakeStatus();
+			while (status != HandshakeStatus.FINISHED) {
+				writeEncodedIfAny();
+				status = runTasksIfNeeded(result);
+				if (status == HandshakeStatus.NEED_UNWRAP) {
+					status = waitForHandshakeData(result, status);
+				}
+				if (status == HandshakeStatus.NEED_WRAP ||
+						status == HandshakeStatus.NOT_HANDSHAKING ||
+						status == HandshakeStatus.FINISHED) {
+					result = encode(plainText);
+					status = result.getHandshakeStatus();
+					if (status == HandshakeStatus.NOT_HANDSHAKING ||
+						status == HandshakeStatus.FINISHED) {
+						break;
+					}
+				}
+				else {
+					logger.debug(status);
+				}
+			}
+		}
+
+		private void writeEncodedIfAny() throws IOException {
+			TcpNioSSLConnection.this.encoded.flip();
+			writeEncoded(TcpNioSSLConnection.this.encoded);
+			TcpNioSSLConnection.this.encoded.clear();
+		}
+
+		/**
+		 * Suspend processing until data is received from the peer.
+		 */
+		private HandshakeStatus waitForHandshakeData(SSLEngineResult result,
+				HandshakeStatus status) throws IOException {
+			try {
+				if (logger.isTraceEnabled()) {
+					logger.trace("Writer waiting for handshake");
+				}
+				if (!semaphore.tryAcquire(30, TimeUnit.SECONDS)) {
+					throw new MessagingException("SSL Handshaking taking too long");
+				}
+				if (logger.isTraceEnabled()) {
+					logger.trace("Writer resuming handshake");
+				}
+				status = runTasksIfNeeded(result);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new MessagingException("Interrupted during SSL Handshaking");
+			}
+			return status;
+		}
+
+		/**
+		 * Encrypts plain text data. The result may indicate handshaking is needed.
+		 */
+		private SSLEngineResult encode(ByteBuffer plainText)
+				throws SSLException, IOException {
+			TcpNioSSLConnection.this.encoded.clear();
+			SSLEngineResult result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded);
+			if (logger.isDebugEnabled()) {
+				logger.debug("After wrap:" + resultToString(result) + " Plaintext buffer @" + plainText.position() + "/" + plainText.limit());
+			}
+			if (result.getStatus() == SSLEngineResult.Status.BUFFER_OVERFLOW) {
+				TcpNioSSLConnection.this.encoded = allocateEncryptionBuffer(sslEngine.getSession().getPacketBufferSize());
+				result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded);
+			}
+			return result;
+		}
+
+		/**
+		 * Write data to the SocketChannel.
+		 */
+		void writeEncoded(ByteBuffer encoded) throws IOException {
+			this.channelOutputStream.doWrite(encoded);
+		}
+	}
+}

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

@@ -30,6 +30,10 @@ import java.nio.channels.SocketChannel;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioConnectionSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpNioConnectionSupport;
+import org.springframework.util.Assert;
+
 /**
 /**
  * Implements a server connection factory that produces {@link TcpNioConnection}s using
@@ -40,14 +44,16 @@ import java.util.Map;
  */
 public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory {
 
-	private ServerSocketChannel serverChannel;
-	
-	private boolean usingDirectBuffers;
-	
-	private Map<SocketChannel, TcpNioConnection> channelMap = new HashMap<SocketChannel, TcpNioConnection>();
+	private volatile ServerSocketChannel serverChannel;
+
+	private volatile boolean usingDirectBuffers;
+
+	private final Map<SocketChannel, TcpNioConnection> channelMap = new HashMap<SocketChannel, TcpNioConnection>();
+
+	private volatile Selector selector;
+
+	private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport();
 
-	private Selector selector;
-	
 	/**
 	 * Listens for incoming connections on the port.
 	 * @param port The port.
@@ -71,7 +77,10 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 		try {
 			this.serverChannel = ServerSocketChannel.open();
 			int port = this.getPort();
-			logger.info("Listening on port " + port);
+			this.getTcpSocketSupport().postProcessServerSocket(this.serverChannel.socket());
+			if (logger.isInfoEnabled()) {
+				logger.info("Listening on port " + port);
+			}
 			this.serverChannel.configureBlocking(false);
 			if (this.getLocalAddress() == null) {
 				this.serverChannel.socket().bind(new InetSocketAddress(port),
@@ -154,7 +163,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 
 	private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {
 		try {
-			TcpNioConnection connection = new TcpNioConnection(socketChannel, true, this.isLookupHost());
+			TcpNioConnection connection = this.tcpNioConnectionSupport
+					.createNewConnection(socketChannel, true,
+							this.isLookupHost());
 			connection.setUsingDirectBuffers(this.usingDirectBuffers);
 			TcpConnection wrappedConnection = wrapConnection(connection);
 			this.initializeConnection(wrappedConnection, socketChannel.socket());
@@ -182,6 +193,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 		this.usingDirectBuffers = usingDirectBuffers;
 	}
 
+	public void setTcpNioConnectionSupport(TcpNioConnectionSupport tcpNioSupport) {
+		Assert.notNull(tcpNioSupport, "TcpNioSupport must not be null");
+		this.tcpNioConnectionSupport = tcpNioSupport;
+	}
+
 	/**
 	 * @return the serverChannel
 	 */
@@ -202,6 +218,5 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	protected Map<SocketChannel, TcpNioConnection> getConnections() {
 		return channelMap;
 	}
-	
 
 }

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

@@ -0,0 +1,58 @@
+/*
+ * 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.support;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLContext;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of TcpSocketFactorySupport
+ * for SSL sockets {@link javax.net.ssl.SSLServerSocket} and
+ * {@link javax.net.ssl.SSLSocket}.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class DefaultTcpNetSSLSocketFactorySupport implements TcpSocketFactorySupport,
+		InitializingBean {
+
+	private final TcpSSLContextSupport sslContextSupport;
+
+	private volatile SSLContext sslContext;
+
+	public DefaultTcpNetSSLSocketFactorySupport(TcpSSLContextSupport sslContextSupport) {
+		Assert.notNull(sslContextSupport, "TcpSSLContextSupport must not be null");
+		this.sslContextSupport = sslContextSupport;
+	}
+
+	public ServerSocketFactory getServerSocketFactory() {
+		return this.sslContext.getServerSocketFactory();
+	}
+
+	public SocketFactory getSocketFactory() {
+		return this.sslContext.getSocketFactory();
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		this.sslContext = this.sslContextSupport.getSSLContext();
+		Assert.notNull(this.sslContext, "SSLContex must not be null");
+	}
+
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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.support;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+
+/**
+ * Implementation of TcpSocketFactorySupport
+ * for non-SSL sockets {@link java.net.ServerSocket} and
+ * {@link java.net.Socket}.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class DefaultTcpNetSocketFactorySupport implements TcpSocketFactorySupport {
+
+	public ServerSocketFactory getServerSocketFactory() {
+		return ServerSocketFactory.getDefault();
+	}
+
+	public SocketFactory getSocketFactory() {
+		return SocketFactory.getDefault();
+	}
+
+}

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

@@ -0,0 +1,36 @@
+/*
+ * 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.support;
+
+import java.nio.channels.SocketChannel;
+
+import org.springframework.integration.ip.tcp.connection.TcpNioConnection;
+
+/**
+ * Implementation of {@link TcpNioConnectionSupport} for non-SSL
+ * NIO connections.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class DefaultTcpNioConnectionSupport implements TcpNioConnectionSupport {
+
+	public TcpNioConnection createNewConnection(SocketChannel socketChannel,
+			boolean server, boolean lookupHost) throws Exception {
+		return new TcpNioConnection(socketChannel, server, lookupHost);
+	}
+
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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.support;
+
+import java.nio.channels.SocketChannel;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.ip.tcp.connection.TcpNioConnection;
+import org.springframework.integration.ip.tcp.connection.TcpNioSSLConnection;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link TcpNioConnectionSupport} for SSL
+ * NIO connections.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class DefaultTcpNioSSLConnectionSupport implements TcpNioConnectionSupport, InitializingBean {
+
+	private volatile SSLContext sslContext;
+
+	private final TcpSSLContextSupport sslContextSupport;
+
+	public DefaultTcpNioSSLConnectionSupport(TcpSSLContextSupport sslContextSupport) {
+		Assert.notNull(sslContextSupport, "TcpSSLContextSupport must not be null");
+		this.sslContextSupport = sslContextSupport;
+	}
+
+	public TcpNioConnection createNewConnection(SocketChannel socketChannel,
+			boolean server, boolean lookupHost) throws Exception {
+
+		SSLEngine sslEngine = this.sslContext.createSSLEngine();
+		TcpNioSSLConnection tcpNioSSLConnection = new TcpNioSSLConnection(socketChannel, server, lookupHost, sslEngine);
+		tcpNioSSLConnection.init();
+		return tcpNioSSLConnection;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		this.sslContext = this.sslContextSupport.getSSLContext();
+		Assert.notNull(this.sslContext, "SSLContex must not be null");
+	}
+
+}

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

@@ -0,0 +1,97 @@
+/*
+ * 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.support;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.util.Assert;
+
+/**
+ * Default implementation of {@link TcpSSLContextSupport}; uses a
+ * 'TLS' (by default) {@link SSLContext}, initialized with 'JKS'
+ * keystores, managed by 'SunX509' Key and Trust managers.
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+public class DefaultTcpSSLContextSupport implements TcpSSLContextSupport {
+
+	private final String keyStore;
+
+	private final String trustStore;
+
+	private final char[] keyStorePassword;
+
+	private final char[] trustStorePassword;
+
+	private volatile String protocol = "TLS";
+
+	/**
+	 * Prepares for the creation of an SSLContext using the supplied
+	 * key/trust stores and passwords.
+	 * @param keyStore A {@link Resource} pointing to the keyStore.
+	 * @param trustStore A {@link Resource} pointing to the trustStore.
+	 * @param keyStorePassword The passowrd for the keyStore.
+	 * @param trustStorePassword The password for the trustStore.
+	 */
+	public DefaultTcpSSLContextSupport(String keyStore, String trustStore,
+			String keyStorePassword, String trustStorePassword) {
+		this.keyStore = keyStore;
+		this.trustStore = trustStore;
+		this.keyStorePassword = keyStorePassword.toCharArray();
+		this.trustStorePassword = trustStorePassword.toCharArray();
+	}
+
+	public SSLContext getSSLContext() throws GeneralSecurityException, IOException  {
+		KeyStore ks = KeyStore.getInstance("JKS");
+		KeyStore ts = KeyStore.getInstance("JKS");
+
+		ks.load(new FileInputStream(new ClassPathResource(keyStore).getFile()), keyStorePassword);
+		ts.load(new FileInputStream(new ClassPathResource(trustStore).getFile()), trustStorePassword);
+
+		KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
+		kmf.init(ks, keyStorePassword);
+
+		TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
+		tmf.init(ts);
+
+		SSLContext sslContext = SSLContext.getInstance(protocol);
+
+		sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+
+		return sslContext;
+
+	}
+
+	/**
+	 * The protocol used in {@link SSLContext#getInstance(String)}; default "TLS".
+	 * @param protocol The protocol.
+	 */
+	public void setProtocol(String protocol) {
+		Assert.notNull(protocol, "protocol must not be null");
+		this.protocol = protocol;
+	}
+
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.support;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+/**
+ * Default implementation of {@link TcpSocketSupport}; makes no
+ * changes to sockets.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class DefaultTcpSocketSupport implements TcpSocketSupport {
+
+	/**
+	 * No-Op.
+	 */
+	public void postProcessServerSocket(ServerSocket serverSocket) {
+	}
+
+	/**
+	 * No-Op.
+	 */
+	public void postProcessSocket(Socket socket) {
+	}
+
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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.support;
+
+import java.nio.channels.SocketChannel;
+
+import org.springframework.integration.ip.tcp.connection.TcpNioConnection;
+
+/**
+ * Used by NIO connection factories to instantiate a {@link TcpNioConnection} object.
+ * Implementations for SSL and non-SSL {@link TcpNioConnection}s are provided.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public interface TcpNioConnectionSupport {
+
+	TcpNioConnection createNewConnection(SocketChannel socketChannel,
+			boolean server, boolean lookupHost) throws Exception;
+
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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.support;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+
+import javax.net.ssl.SSLContext;
+
+/**
+ * Strategy interface for the creation of an {@link SSLContext} object
+ * for use with SSL/TLS sockets.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public interface TcpSSLContextSupport {
+
+	/**
+	 * Gets an SSLContext.
+	 * @return the SSLContext.
+	 * @throws Exception
+	 */
+	SSLContext getSSLContext() throws GeneralSecurityException, IOException;
+
+}

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

@@ -0,0 +1,46 @@
+/*
+ * 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.support;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+
+/**
+ * Strategy interface for supplying Socket Factories.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public interface TcpSocketFactorySupport {
+
+	/**
+	 * Supplies the {@link ServerSocketFactory} to be used to
+	 * create new {@link ServerSocket}s.
+	 * @return the ServerSocketFacory
+	 */
+    ServerSocketFactory getServerSocketFactory();
+
+    /**
+     * Supplies the {@link SocketFactory} to be used to
+     * create new {@link Socket}s.
+     * @return the SocketFactory
+     */
+    SocketFactory getSocketFactory();
+
+}

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

@@ -0,0 +1,47 @@
+/*
+ * 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.support;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+/**
+ * Strategy interface for modifying sockets.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public interface TcpSocketSupport {
+
+	/**
+	 * Performs any further modifications to the server socket
+	 * after the connection factory has created the socket and
+	 * set any configured attributes, before invoking
+	 * {@link ServerSocket#accept()}.
+	 * @param serverSocket The ServerSocket
+	 */
+    void postProcessServerSocket(ServerSocket serverSocket);
+
+    /**
+     * Performs any further modifications to the {@link Socket} after
+     * the socket has been created by a client, or accepted by
+     * a server, and after any configured atributes have been
+     * set.
+     * @param socket The Socket
+     */
+    void postProcessSocket(Socket socket);
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides classes supporting the creation/manipulation of sockets,
+ * SSLContexts etc.
+ */
+package org.springframework.integration.ip.tcp.connection.support;

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

@@ -22,19 +22,21 @@ import java.io.OutputStream;
 
 /**
  * Reads data in an InputStream to a byte[]; data must be terminated by \r\n
- * (not included in resulting byte[]). 
+ * (not included in resulting byte[]).
  * Writes a byte[] to an OutputStream and adds \r\n.
- * 
+ *
  * @author Gary Russell
  * @since 2.0
  */
 public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer {
 
+	private static final byte[] CRLF = "\r\n".getBytes();
+
 	/**
 	 * Reads the data in the inputstream to a byte[]. Data must be terminated
 	 * by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream
 	 * is closed immediately after the \r\n (i.e. no data is in the process of
-	 * being read). 
+	 * being read).
 	 */
 	public byte[] deserialize(InputStream inputStream) throws IOException {
 		byte[] buffer = new byte[this.maxMessageSize];
@@ -69,8 +71,7 @@ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer {
 	 */
 	public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
 		outputStream.write(bytes);
-		outputStream.write('\r');
-		outputStream.write('\n');
+		outputStream.write(CRLF);
 		outputStream.flush();
 	}
 

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

@@ -491,6 +491,52 @@ connections created by this factory. Facilitates resequencing if necessary. Defa
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="ssl-context-support" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+A reference to a TcpSSLContextSupport strategy implementation. Providing this reference
+enables SSL on connections created by this factory. A DefaultTcpSSLContextSupport implementation
+is provided that takes keystore and trustore names and passwords. The SSLContext created by
+this implementation is used to obtain socket factories (when using-nio="false") or SSLEngine
+instances (when using-nio="true"). When this attribute is omitted, normal plain text
+sockets are used.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="socket-support" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+A reference to a TcpSocketSupport strategy implementation. Allows post-processing Socket
+and ServerSocket instances after creation and after configured attributes are applied.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="socket-factory-support" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+A reference to a TcpSocketFactorySupport strategy implementation. Allows customization
+of the factories used to create sockets. The default implementation returns default.
+ServerSocketFactory and SocketFactory instances, unless an 'ssl-context-support'
+attribute has been supplied, in which case the SSLContext obtained therefrom is
+used to create SSLServerSocketFactory and SSLSocketFactory instances.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 

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

@@ -58,8 +58,33 @@
 		port="#{tcpIpUtils.findAvailableServerSocket(5200)}"
 		lookup-host="false"
 		apply-sequence="true"
+		ssl-context-support="sslContextSupport"
 		/>
 
+	<bean id="sslContextSupport" class="org.springframework.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport">
+		<constructor-arg value="test.ks"/>
+		<constructor-arg value="test.truststore.ks"/>
+		<constructor-arg value="secret"/>
+		<constructor-arg value="secret"/>
+	</bean>
+
+	<ip:tcp-connection-factory id="secureServer"
+		type="server"
+		port="#{tcpIpUtils.findAvailableServerSocket(5250)}"
+		lookup-host="false"
+		apply-sequence="true"
+		ssl-context-support="sslContextSupport"
+		socket-support="socketSupport"
+		socket-factory-support="socketFactorySupport" />
+
+	<bean id="socketSupport" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport" />
+	</bean>
+
+	<bean id="socketFactorySupport" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport" />
+	</bean>
+
 	<ip:tcp-inbound-channel-adapter id="testInTcp"
 		channel="tcpChannel"
 		error-channel="errorChannel"

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

@@ -29,7 +29,6 @@ import java.util.Set;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -50,6 +49,9 @@ import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionF
 import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSSLSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport;
 import org.springframework.integration.ip.udp.DatagramPacketMessageMapper;
 import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
 import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
@@ -212,6 +214,15 @@ public class ParserUnitTests {
 	@Autowired @Qualifier("udpAutoChannel.adapter")
 	UnicastReceivingChannelAdapter udpAutoAdapter;
 
+	@Autowired
+	TcpNetServerConnectionFactory secureServer;
+
+	@Autowired
+	TcpSocketFactorySupport socketFactorySupport;
+
+	@Autowired
+	TcpSocketSupport socketSupport;
+
 	@Test
 	public void testInUdp() {
 		DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn);
@@ -261,6 +272,7 @@ public class ParserUnitTests {
 		assertEquals(124, tcpIn.getPhase());
 		assertTrue((Boolean) TestUtils.getPropertyValue(
 				TestUtils.getPropertyValue(cfS1, "mapper"), "applySequence"));
+		assertTrue(TestUtils.getPropertyValue(cfS1, "tcpSocketFactorySupport") instanceof DefaultTcpNetSSLSocketFactorySupport);
 	}
 
 	@Test
@@ -544,4 +556,11 @@ public class ParserUnitTests {
 	public void testAutoUdp() {
 		assertSame(udpAutoChannel, TestUtils.getPropertyValue(udpAutoAdapter, "outputChannel"));
 	}
+
+	@Test
+	public void testSecureServer() {
+		DirectFieldAccessor dfa = new DirectFieldAccessor(secureServer);
+		assertSame(socketFactorySupport, dfa.getPropertyValue("tcpSocketFactorySupport"));
+		assertSame(socketSupport, dfa.getPropertyValue("tcpSocketSupport"));
+	}
 }

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

@@ -0,0 +1,460 @@
+/*
+ * 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.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLEngine;
+
+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.ip.tcp.connection.support.DefaultTcpNetSSLSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioSSLConnectionSupport;
+import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport;
+import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+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 SocketSupportTests {
+
+	@Test
+	public void testNetClient() throws Exception {
+		TcpSocketFactorySupport factorySupport = mock(TcpSocketFactorySupport.class);
+		SocketFactory factory = Mockito.mock(SocketFactory.class);
+		when(factorySupport.getSocketFactory()).thenReturn(factory);
+		Socket socket = mock(Socket.class);
+		InputStream is = mock(InputStream.class);
+		when(is.read()).thenReturn(-1);
+		when(socket.getInputStream()).thenReturn(is);
+		InetAddress inetAddress = InetAddress.getLocalHost();
+		when(socket.getInetAddress()).thenReturn(inetAddress);
+		when(factory.createSocket("x", 0)).thenReturn(socket);
+		TcpSocketSupport socketSupport = Mockito.mock(TcpSocketSupport.class);
+
+		TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory("x", 0);
+		connectionFactory.setTcpSocketFactorySupport(factorySupport);
+		connectionFactory.setTcpSocketSupport(socketSupport);
+		connectionFactory.start();
+		connectionFactory.getConnection();
+
+		verify(socketSupport).postProcessSocket(socket);
+		connectionFactory.stop();
+	}
+
+	@Test
+	public void testNetServer() throws Exception {
+		TcpSocketFactorySupport factorySupport = mock(TcpSocketFactorySupport.class);
+		ServerSocketFactory factory = mock(ServerSocketFactory.class);
+		when(factorySupport.getServerSocketFactory()).thenReturn(factory);
+		Socket socket = mock(Socket.class);
+		InputStream is = mock(InputStream.class);
+		when(is.read()).thenReturn(-1);
+		when(socket.getInputStream()).thenReturn(is);
+		InetAddress inetAddress = InetAddress.getLocalHost();
+		when(socket.getInetAddress()).thenReturn(inetAddress);
+		ServerSocket serverSocket = mock(ServerSocket.class);
+		when(serverSocket.getInetAddress()).thenReturn(inetAddress);
+		when(factory.createServerSocket(0, 5)).thenReturn(serverSocket);
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		when(serverSocket.accept()).thenReturn(socket).then(new Answer<Socket> (){
+			public Socket answer(InvocationOnMock invocation) throws Throwable {
+				latch1.countDown();
+				latch2.await(10, TimeUnit.SECONDS);
+				return null;
+			}});
+		TcpSocketSupport socketSupport = mock(TcpSocketSupport.class);
+
+		TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(0);
+		connectionFactory.setTcpSocketFactorySupport(factorySupport);
+		connectionFactory.setTcpSocketSupport(socketSupport);
+		connectionFactory.registerListener(mock(TcpListener.class));
+		connectionFactory.start();
+
+		assertTrue(latch1.await(10, TimeUnit.SECONDS));
+		verify(socketSupport).postProcessServerSocket(serverSocket);
+		verify(socketSupport).postProcessSocket(socket);
+		latch2.countDown();
+		connectionFactory.stop();
+	}
+
+	@Test
+	public void testNioClientAndServer() throws Exception {
+		int port = SocketTestUtils.findAvailableServerSocket();
+		TcpNioClientConnectionFactory clientConnectionFactory = new TcpNioClientConnectionFactory("localhost", port);
+		final AtomicInteger ppSocketCountClient = new AtomicInteger();
+		final AtomicInteger ppServerSocketCountClient = new AtomicInteger();
+		TcpSocketSupport clientSocketSupport = new TcpSocketSupport() {
+			public void postProcessSocket(Socket socket) {
+				ppSocketCountClient.incrementAndGet();
+			}
+			public void postProcessServerSocket(ServerSocket serverSocket) {
+				ppServerSocketCountClient.incrementAndGet();
+			}
+		};
+		clientConnectionFactory.setTcpSocketSupport(clientSocketSupport);
+		clientConnectionFactory.start();
+		TcpNioServerConnectionFactory serverConnectionFactory = new TcpNioServerConnectionFactory(port);
+		serverConnectionFactory.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				return false;
+			}
+		});
+		final AtomicInteger ppSocketCountServer = new AtomicInteger();
+		final AtomicInteger ppServerSocketCountServer = new AtomicInteger();
+		TcpSocketSupport serverSocketSupport = new TcpSocketSupport() {
+			public void postProcessSocket(Socket socket) {
+				ppSocketCountServer.incrementAndGet();
+			}
+			public void postProcessServerSocket(ServerSocket serverSocket) {
+				ppServerSocketCountServer.incrementAndGet();
+			}
+		};
+		serverConnectionFactory.setTcpSocketSupport(serverSocketSupport);
+		serverConnectionFactory.start();
+		waitListening(serverConnectionFactory);
+		clientConnectionFactory.getConnection().send(new GenericMessage<String>("Hello, world!"));
+		assertEquals(0, ppServerSocketCountClient.get());
+		assertEquals(1, ppSocketCountClient.get());
+
+		assertEquals(1, ppServerSocketCountServer.get());
+		assertEquals(1, ppSocketCountServer.get());
+	}
+
+	/*
+$ keytool -genkeypair -alias sitestcertkey -keyalg RSA -validity 36500 -keystore src/test/resources/test.ks
+Enter keystore password: secret
+Re-enter new password: secret
+What is your first and last name?
+  [Unknown]:  Spring Integration
+What is the name of your organizational unit?
+  [Unknown]:  SpringSource
+What is the name of your organization?
+  [Unknown]:  VMware
+What is the name of your City or Locality?
+  [Unknown]:  Palo Alto
+What is the name of your State or Province?
+  [Unknown]:  CA
+What is the two-letter country code for this unit?
+  [Unknown]:  US
+Is CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US correct?
+  [no]:  yes
+
+Enter key password for <certificatekey>
+	(RETURN if same as keystore password):
+
+$ keytool -list -v -keystore src/test/resources/test.ks
+Enter keystore password: secret
+
+Keystore type: JKS
+Keystore provider: SUN
+
+Your keystore contains 1 entry
+
+Alias name: sitestcertkey
+Creation date: Feb 25, 2012
+Entry type: PrivateKeyEntry
+Certificate chain length: 1
+Certificate[1]:
+Owner: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US
+Issuer: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US
+Serial number: 4f491902
+Valid from: Sat Feb 25 12:23:14 EST 2012 until: Mon Feb 01 12:23:14 EST 2112
+Certificate fingerprints:
+	 MD5:  4F:A9:76:0E:A9:C0:A8:B7:26:E7:7E:C7:E8:22:1F:8B
+	 SHA1: 88:AC:9E:4D:29:0D:3A:59:3B:73:95:4A:E1:BB:D0:22:89:37:64:4C
+	 Signature algorithm name: SHA1withRSA
+	 Version: 3
+
+
+*******************************************
+*******************************************
+
+$ keytool -export -alias sitestcertkey -keystore src/test/resources/test.ks -rfc -file src/test/resources/test.cer
+Enter keystore password:
+Certificate stored in file <src/test/resources/test.cer>
+
+$ keytool -import -alias sitestcertkey -file src/test/resources/test.cer -keystore src/test/resources/test.truststore.ks
+Enter keystore password: secret
+Re-enter new password: secret
+Owner: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US
+Issuer: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US
+Serial number: 4f491902
+Valid from: Sat Feb 25 12:23:14 EST 2012 until: Mon Feb 01 12:23:14 EST 2112
+Certificate fingerprints:
+	 MD5:  4F:A9:76:0E:A9:C0:A8:B7:26:E7:7E:C7:E8:22:1F:8B
+	 SHA1: 88:AC:9E:4D:29:0D:3A:59:3B:73:95:4A:E1:BB:D0:22:89:37:64:4C
+	 Signature algorithm name: SHA1withRSA
+	 Version: 3
+Trust this certificate? [no]:  yes
+Certificate was added to keystore
+
+$ keytool -list -v -keystore src/test/resources/test.truststore.ks
+Enter keystore password: secret
+
+Keystore type: JKS
+Keystore provider: SUN
+
+Your keystore contains 1 entry
+
+Alias name: sitestcertkey
+Creation date: Feb 25, 2012
+Entry type: trustedCertEntry
+
+Owner: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US
+Issuer: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US
+Serial number: 4f491902
+Valid from: Sat Feb 25 12:23:14 EST 2012 until: Mon Feb 01 12:23:14 EST 2112
+Certificate fingerprints:
+	 MD5:  4F:A9:76:0E:A9:C0:A8:B7:26:E7:7E:C7:E8:22:1F:8B
+	 SHA1: 88:AC:9E:4D:29:0D:3A:59:3B:73:95:4A:E1:BB:D0:22:89:37:64:4C
+	 Signature algorithm name: SHA1withRSA
+	 Version: 3
+
+
+*******************************************
+*******************************************
+
+	 */
+	@Test
+	public void testNetClientAndServerSSL() throws Exception {
+		System.setProperty("javax.net.debug", "all"); // SSL activity in the console
+		int port = SocketTestUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		TcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
+				"test.truststore.ks", "secret", "secret");
+		DefaultTcpNetSSLSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(sslContextSupport);
+		tcpSocketFactorySupport.afterPropertiesSet();
+		server.setTcpSocketFactorySupport(tcpSocketFactorySupport);
+		final List<Message<?>> messages = new ArrayList<Message<?>>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				messages.add(message);
+				latch.countDown();
+				return false;
+			}
+		});
+		server.start();
+		waitListening(server);
+
+		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
+		client.setTcpSocketFactorySupport(tcpSocketFactorySupport);
+		client.start();
+
+		TcpConnection connection = client.getConnection();
+		connection.send(new GenericMessage<String>("Hello, world!"));
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload()));
+	}
+
+	@Test
+	public void testNetClientAndServerSSLDifferentContexts() throws Exception {
+		System.setProperty("javax.net.debug", "all"); // SSL activity in the console
+		int port = SocketTestUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		TcpSSLContextSupport serverSslContextSupport = new DefaultTcpSSLContextSupport("server.ks",
+				"server.truststore.ks", "secret", "secret");
+		DefaultTcpNetSSLSocketFactorySupport serverTcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(serverSslContextSupport);
+		serverTcpSocketFactorySupport.afterPropertiesSet();
+		server.setTcpSocketFactorySupport(serverTcpSocketFactorySupport);
+		final List<Message<?>> messages = new ArrayList<Message<?>>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				messages.add(message);
+				latch.countDown();
+				return false;
+			}
+		});
+		server.start();
+		waitListening(server);
+
+		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
+		TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks", "client.truststore.ks",
+				"secret", "secret");
+		DefaultTcpNetSSLSocketFactorySupport clientTcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(clientSslContextSupport);
+		clientTcpSocketFactorySupport.afterPropertiesSet();
+		client.setTcpSocketFactorySupport(clientTcpSocketFactorySupport);
+		client.start();
+
+		TcpConnection connection = client.getConnection();
+		connection.send(new GenericMessage<String>("Hello, world!"));
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload()));
+	}
+
+	@Test
+	public void testNioClientAndServerSSL() throws Exception {
+		System.setProperty("javax.net.debug", "all"); // SSL activity in the console
+		int port = SocketTestUtils.findAvailableServerSocket();
+		TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(port);
+		DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
+				"test.truststore.ks", "secret", "secret");
+		sslContextSupport.setProtocol("SSL");
+		DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(sslContextSupport);
+		tcpNioConnectionSupport.afterPropertiesSet();
+		server.setTcpNioConnectionSupport(tcpNioConnectionSupport);
+		final List<Message<?>> messages = new ArrayList<Message<?>>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				System.out.println("Server" + message);
+				messages.add(message);
+				latch.countDown();
+				return false;
+			}
+		});
+		server.start();
+		waitListening(server);
+
+		TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
+		client.setTcpNioConnectionSupport(tcpNioConnectionSupport);
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				System.out.println("Client" + message);
+				return false;
+			}
+		});
+		client.start();
+
+		TcpConnection connection = client.getConnection();
+		connection.send(new GenericMessage<String>("Hello, world!"));
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload()));
+	}
+
+	@Test
+	public void testNioClientAndServerSSLDifferentContextsLargeDataWithReply() throws Exception {
+		System.setProperty("javax.net.debug", "all"); // SSL activity in the console
+		int port = SocketTestUtils.findAvailableServerSocket();
+		TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(port);
+		TcpSSLContextSupport serverSslContextSupport = new DefaultTcpSSLContextSupport("server.ks",
+				"server.truststore.ks", "secret", "secret");
+		DefaultTcpNioSSLConnectionSupport serverTcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport);
+		serverTcpNioConnectionSupport.afterPropertiesSet();
+		server.setTcpNioConnectionSupport(serverTcpNioConnectionSupport);
+		final List<Message<?>> messages = new ArrayList<Message<?>>();
+		final CountDownLatch latch = new CountDownLatch(2);
+		final Replier replier = new Replier();
+		server.registerSender(replier);
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				System.out.println("Server:" + message);
+				messages.add(message);
+				try {
+					replier.send(message);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+				latch.countDown();
+				return false;
+			}
+		});
+		ByteArrayCrLfSerializer deserializer = new ByteArrayCrLfSerializer();
+		deserializer.setMaxMessageSize(120000);
+		server.setDeserializer(deserializer);
+		server.start();
+		waitListening(server);
+
+		TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
+		TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks",
+				"client.truststore.ks", "secret", "secret");
+		DefaultTcpNioSSLConnectionSupport clientTcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(clientSslContextSupport);
+		clientTcpNioConnectionSupport.afterPropertiesSet();
+		client.setTcpNioConnectionSupport(clientTcpNioConnectionSupport);
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				System.out.println("Client:" + message);
+				messages.add(message);
+				latch.countDown();
+				return false;
+			}
+		});
+		client.setDeserializer(deserializer);
+		client.start();
+
+		TcpConnection connection = client.getConnection();
+		byte[] bytes = new byte[100000];
+		connection.send(new GenericMessage<String>("Hello, world!" + new String(bytes)));
+		assertTrue(latch.await(60, TimeUnit.SECONDS));
+		byte[] payload = (byte[]) messages.get(0).getPayload();
+		assertEquals(13 + bytes.length, payload.length);
+		assertEquals("Hello, world!", new String(payload).substring(0, 13));
+		payload = (byte[]) messages.get(1).getPayload();
+		assertEquals(13 + bytes.length, payload.length);
+		assertEquals("Hello, world!", new String(payload).substring(0, 13));
+	}
+
+	private void waitListening(AbstractServerConnectionFactory scf) throws Exception {
+		int n = 0;
+		while (!scf.isListening()) {
+			Thread.sleep(100);
+			if (++n > 100) {
+				fail("Server failed to start listening");
+			}
+		}
+	}
+
+	private class Replier implements TcpSender {
+
+		private TcpConnection connection;
+
+		public void addNewConnection(TcpConnection connection) {
+			this.connection = connection;
+		}
+
+		public void removeDeadConnection(TcpConnection connection) {
+		}
+
+		public void send(Message<?> message) throws Exception {
+			// force a renegotiation from the server side
+			SSLEngine sslEngine = TestUtils.getPropertyValue(this.connection, "sslEngine", SSLEngine.class);
+			sslEngine.getSession().invalidate();
+			sslEngine.beginHandshake();
+			this.connection.send(message);
+		}
+	}
+}

spring-integration-ip/src/test/resources/client.cer

@@ -0,0 +1,14 @@
+-----BEGIN CERTIFICATE-----
+MIICdzCCAeCgAwIBAgIET0k29DANBgkqhkiG9w0BAQUFADB/MQswCQYDVQQGEwJVUzELMAkGA1UE
+CBMCQ0ExEjAQBgNVBAcTCVBhbG8gQWx0bzEPMA0GA1UEChMGVk13YXJlMRUwEwYDVQQLEwxTcHJp
+bmdTb3VyY2UxJzAlBgNVBAMTHlNwcmluZyBJbnRlZ3JhdGlvbiBUZXN0IENsaWVudDAgFw0xMjAy
+MjUxOTMxMDBaGA8yMTEyMDIwMTE5MzEwMFowfzELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAkNBMRIw
+EAYDVQQHEwlQYWxvIEFsdG8xDzANBgNVBAoTBlZNd2FyZTEVMBMGA1UECxMMU3ByaW5nU291cmNl
+MScwJQYDVQQDEx5TcHJpbmcgSW50ZWdyYXRpb24gVGVzdCBDbGllbnQwgZ8wDQYJKoZIhvcNAQEB
+BQADgY0AMIGJAoGBAKL9XxG1vf6SXW7iNKR7bMbEqIgWzg6rxsr1e+35zsmTHhGwoRJJw/Fm/+Sc
+bHxkO3YzM+WayQwe0NXttjMgjLnFoztEpTpWYsmzJt0r6fv7rGtixXFfLjmulONmakGFptwRZlWv
+HhdDYkQiw3P8npC/ZWMQagaMHpxzlzyeD6BfAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAbsPnIYVl
+1ClEkmPp/aSZooGtEd1JXUvW4JXWWt6oxTVe2SdU62GvxHDc71z73Fvqk1BFoKtniuOgrjZNeL2j
+kGVbeB2wT+aG9soXifcYolNDXeNVF5xpYxHmDb1LyMiEpHl071mhD0q+a5Q1gFly5HbDb+RvrsNt
+k6ooEx5Ycuo=
+-----END CERTIFICATE-----

spring-integration-ip/src/test/resources/server.cer

@@ -0,0 +1,14 @@
+-----BEGIN CERTIFICATE-----
+MIICdzCCAeCgAwIBAgIET0k3GjANBgkqhkiG9w0BAQUFADB/MQswCQYDVQQGEwJVUzELMAkGA1UE
+CBMCQ0ExEjAQBgNVBAcTCVBhbG8gQWx0bzEPMA0GA1UEChMGVk13YXJlMRUwEwYDVQQLEwxTcHJp
+bmdTb3VyY2UxJzAlBgNVBAMTHlNwcmluZyBJbnRlZ3JhdGlvbiBUZXN0IFNlcnZlcjAgFw0xMjAy
+MjUxOTMxMzhaGA8yMTEyMDIwMTE5MzEzOFowfzELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAkNBMRIw
+EAYDVQQHEwlQYWxvIEFsdG8xDzANBgNVBAoTBlZNd2FyZTEVMBMGA1UECxMMU3ByaW5nU291cmNl
+MScwJQYDVQQDEx5TcHJpbmcgSW50ZWdyYXRpb24gVGVzdCBTZXJ2ZXIwgZ8wDQYJKoZIhvcNAQEB
+BQADgY0AMIGJAoGBAM/YT0knf57/elITdhqQ5TgUKcbIKFncb3tmDrbZTm8kKWT21cUUvx5g0ZH4
+dHtCDOY1Zjd4Qyh0YQ1tXbpccpU014IKDHAc85BQGoUiFpUKAMrJ7jGff7iflbeXMgizFnQXtdb/
+fH7E8YGTP0XV2RizMPAClWcJoHBf7LjYzbknAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEArGlIeCKf
+Mj+IL9tJU4fxEGTTMCINBBOQB5sOmvsfZZddLcHGoYrFxxZxqUeVwv/VHRAGUe5kN6nTsGn5hFud
+HeFswEgx/YIQtdRW1HYt3uuyZh+20cOq4uaNS7YFvPCIe7yBscK7PRwKrVspVZQFLFtcRIJUkF+0
+cOoQRaygXys=
+-----END CERTIFICATE-----

spring-integration-ip/src/test/resources/test.cer

@@ -0,0 +1,13 @@
+-----BEGIN CERTIFICATE-----
+MIICXzCCAcigAwIBAgIET0kZAjANBgkqhkiG9w0BAQUFADBzMQswCQYDVQQGEwJVUzELMAkGA1UE
+CBMCQ0ExEjAQBgNVBAcTCVBhbG8gQWx0bzEPMA0GA1UEChMGVk13YXJlMRUwEwYDVQQLEwxTcHJp
+bmdTb3VyY2UxGzAZBgNVBAMTElNwcmluZyBJbnRlZ3JhdGlvbjAgFw0xMjAyMjUxNzIzMTRaGA8y
+MTEyMDIwMTE3MjMxNFowczELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAkNBMRIwEAYDVQQHEwlQYWxv
+IEFsdG8xDzANBgNVBAoTBlZNd2FyZTEVMBMGA1UECxMMU3ByaW5nU291cmNlMRswGQYDVQQDExJT
+cHJpbmcgSW50ZWdyYXRpb24wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAM6hHqm4jCixwNgK
+z5kBxsWbuGvSSLMiG8fMbg6RbVmbhh4ssVttzjcC3G2OxUxC2gQ9H/96PwgGJZp4VKZw8cPYVTZe
+kX79NKvv1IBQ661LbFMF7yH0bMNtU8I/dT5P+hrvNbWT/oo5YYvI4LkDfrw4l4lqWNcW5Wyg40NO
+7Yo7AgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAbSkOrZKZ9caK4TJhJPD/6HC8PfJRcRc4hBdM54UX
+4BxW9VRhrjZLS9luWrnVqfrqiZ49UuApTK+5K12GAcmkZGLJzDzaM6D55dW6JC7YlZEQQxHN0GvG
+PqgOxu248fIqrasq4KXUGLvhL31ylRXZIcfEo15XpWwIhrKOWo2MBYw=
+-----END CERTIFICATE-----

src/reference/docbook/ip.xml

@@ -351,6 +351,13 @@
             </para>
         </note>
     </para>
+    <note>
+     <para>
+      It is possible to modify the creation of and/or attributes of sockets - see
+      <xref linkend="ssl-tls"/>. As is noted there, such modifications are possible whether
+      or not SSL is being used.
+     </para>
+    </note>
   </section>
   <section id="ip-interceptors">
     <title>TCP Connection Interceptors</title>
@@ -816,6 +823,157 @@
     a new message is received.
    </para>
   </section>
+  <section id="ssl-tls">
+    <title>SSL/TLS Support</title>
+    <section>
+      <title>Overview</title>
+      <para>
+       Secure Sockets Layer/Transport Layer Security is supported. When using NIO, the JDK 5+
+       <classname>SSLEngine</classname> feature is used to handle handshaking after the
+       connection is established. When not using NIO, standard
+       <classname>SSLSocketFactory</classname> and <classname>SSLServerSocketFactory</classname> objects are
+       used to create connections. A number of strategy interfaces are provided to allow
+       significant customization; default implementations of these interfaces provide for
+       the simplest way to get started with secure communications.
+      </para>
+     </section>
+    <section>
+      <title>Getting Started</title>
+      <para>
+       Regardless of whether NIO is being used, you need to configure the
+       <classname>ssl-context-support</classname> attribute on the connection factory.
+       This attribute references a &lt;bean/&gt; definition that describes the location
+       and passwords for the required key stores.
+      </para>
+      <para>
+       SSL/TLS peers require two keystores each; a keystore containing private/public key
+       pairs identifying the peer; a truststore, containing the public keys for peers that
+       are trusted. See the documentation for the <classname>keytool</classname> utility
+       provided with the JDK. The essential steps are
+      </para>
+      <para>
+       <orderedlist>
+        <listitem><para>Create a new key pair and store in a keystore.</para></listitem>
+        <listitem><para>Export the public key.</para></listitem>
+        <listitem><para>Import the public key into the peer's truststore.</para></listitem>
+       </orderedlist>
+      </para>
+      <para>
+       Repeat for the other peer.
+      </para>
+      <note>
+       <para>
+        It is common in test cases to use the same key stores on both peers, but this should
+        be avoided for production.
+       </para>
+      </note>
+      <para>
+       After establishing the key stores, the next step is to indicate their locations to the
+       <classname>TcpSSLContextSupport</classname> bean, and provide a reference to that bean
+       to the connection factory.
+      </para>
+      <para><programlisting language="xml"><![CDATA[    <bean id="sslContextSupport"
+        class="o.sf.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport">
+        <constructor-arg value="client.ks"/>
+        <constructor-arg value="client.truststore.ks"/>
+        <constructor-arg value="secret"/>
+        <constructor-arg value="secret"/>
+    </bean>
+
+    <ip:tcp-connection-factory id="clientFactory"
+        type="client"
+        host="localhost"
+        port="1234"
+        ssl-context-support="sslContextSupport"]]></programlisting>
+      </para>
+      <para>
+       The <classname>DefaulTcpSSLContextSupport</classname> class also has an optional
+       'protocol' property, which can be 'SSL' or 'TLS' (default).
+      </para>
+      <para>
+       The keystore file names (first two constructor arguments) use the Spring <classname>Resource</classname>
+       abstraction; by default the files will be located on the classpath, but this can be overridden by using
+       the <classname>file:</classname> prefix, to find the files on the filesystem instead.
+      </para>
+    </section>
+    <section>
+      <title>Advanced Techniques</title>
+      <para>
+       In many cases, the configuration described above is all that is needed to enable secure
+       communication over TCP/IP. However, a number of strategy interfaces are provided to
+       allow customization and modification of socket factories and sockets.
+      </para>
+      <para>
+       <itemizedlist>
+        <listitem><para><classname>TcpSSLContextSupport</classname></para></listitem>
+        <listitem><para><classname>TcpSocketFactorySupport</classname></para></listitem>
+        <listitem><para><classname>TcpSocketSupport</classname></para></listitem>
+       </itemizedlist>
+      </para>
+      <para><programlisting language="java"><![CDATA[public interface TcpSSLContextSupport {
+
+	SSLContext getSSLContext() throws Exception;
+
+}]]></programlisting>
+      </para>
+      <para>
+       Implementations of this interface are responsible for creating an SSLContext.
+       The sole implementation provided by the framework is the
+       <classname>DefaultTcpSSLContextSupport</classname> described above. If you require
+       different behavior, implement this interface and provide the connection factory with
+       a reference to a bean of your class' implementation.
+      </para>
+      <para><programlisting language="java"><![CDATA[public interface TcpSocketFactorySupport {
+
+    ServerSocketFactory getServerSocketFactory();
+
+    SocketFactory getSocketFactory();
+
+}
+]]></programlisting>
+      </para>
+      <para>
+       Implementations of this interface are responsible for obtaining references to
+       <classname>ServerSocketFactory</classname> and <classname>SocketFactory</classname>.
+       Two implementations are provided; the first is <classname>DefaultTcpNetSocketFactorySupport</classname>
+       for non-SSL sockets (when no 'ssl-context-support' attribute is defined); this simply
+       uses the JDK's default factories. The second implementation is
+       <classname>DefaultTcpNetSSLSocketFactorySupport</classname>; this is used, by default,
+       when an 'ssl-context-support' attribute is defined; it uses the <classname>SSLContext</classname>
+       created by that bean to create the socket factories.
+      </para>
+      <note>
+       <para>
+        This interface only applies if <classname>using-nio</classname> is "false"; socket factories
+        are not used by NIO.
+       </para>
+      </note>
+      <para><programlisting language="java"><![CDATA[public interface TcpSocketSupport {
+
+    void postProcessServerSocket(ServerSocket serverSocket);
+
+    void postProcessSocket(Socket socket);
+
+]]></programlisting>
+      </para>
+      <para>
+       Implementations of this interface can modify sockets after they are created, and after
+       all configured attributes have been applied, but before the sockets are used. This applies
+       whether or not NIO is being used. For example,
+       you could use an implementation of this interface to modify the supported cipher suites on
+       an SSL socket, or you could add a listener that gets notified after SSL handshaking is
+       complete. The sole implementation provided by the framework is the
+       <classname>DefaultTcpSocketSupport</classname> which does not modify the sockets in
+       any way
+      </para>
+      <para>
+       To supply your own implementation of <classname>TcpSocketFactorySupport</classname> or
+       <classname>TcpSocketSupport</classname>, provide the connection factory with references to
+       beans of your custom type using the <classname>socket-factory-support</classname> and
+       <classname>socket-support</classname> attributes, respectively.
+      </para>
+    </section>
+  </section>
   <section id="ip-endpoint-reference">
     <title>IP Configuration Attributes</title>
     <para>
@@ -1020,6 +1178,27 @@
               <entry></entry>
               <entry>See <xref linkend="ip-interceptors"/> </entry>
             </row>
+            <row>
+              <entry>ssl-context-support</entry>
+              <entry>Y</entry>
+              <entry>Y</entry>
+              <entry></entry>
+              <entry>See <xref linkend="ssl-tls"/> </entry>
+            </row>
+            <row>
+              <entry>socket-factory-support</entry>
+              <entry>Y</entry>
+              <entry>Y</entry>
+              <entry></entry>
+              <entry>See <xref linkend="ssl-tls"/> </entry>
+            </row>
+            <row>
+              <entry>socket-support</entry>
+              <entry>Y</entry>
+              <entry>Y</entry>
+              <entry></entry>
+              <entry>See <xref linkend="ssl-tls"/> </entry>
+            </row>
           </tbody>
         </tgroup>
       </table>