INT-2153 SSL Support

Add strategy interfaces
 - obtaining ServerSocketFactory and SocketFactory
 - post processing ServerSockets and Sockets
 - obtaining initialized SSLContext

Provide SSL and non-SSL implementations of the strategies, to
serve up the appropriate socket factories, and do nothing in the
post processing methods.

The postprocessors allow the user to modify sockets after
configured attributes have been applied but before the sockets
are used.

This is particularly useful with SSL in case additional SSL
options need to be applied.

Docs

Polishing JavaDocs

Polishing

Javadocs, polishing

Polishing

Polishing

INT-2153 Polishing

PR Review

White Space

INT-2513 Polishing

Fixed method names.

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>