INT-1340 Add TCP Connection Interceptor Chain

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

@@ -103,6 +103,8 @@ public abstract class IpAdapterParserUtils {
 
 	static final String TCP_CONNECTION_FACTORY = "connection-factory";
 
+	public static final String INTERCEPTOR_FACTORY_CHAIN = "interceptor-factory-chain";
+
 	/**
 	 * Adds a constructor-arg to the provided bean definition builder 
 	 * with the value of the attribute whose name is provided if that

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

@@ -92,6 +92,8 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
 				IpAdapterParserUtils.OUTPUT_CONVERTER);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SINGLE_USE);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 
+				IpAdapterParserUtils.INTERCEPTOR_FACTORY_CHAIN);
 	
 		return builder.getBeanDefinition();
 	}

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

@@ -103,6 +103,7 @@ public class TcpSendingMessageHandler implements MessageHandler, TcpSender {
 		} catch (MessageMappingException e) {
 			// retry - socket may have closed
 			if (e.getCause() instanceof IOException) {
+				logger.debug("Fail on first write attempt", e);
 				doWrite(message);
 			} else {
 				throw e;
@@ -120,13 +121,20 @@ public class TcpSendingMessageHandler implements MessageHandler, TcpSender {
 			if (connection == null) {
 				throw new MessageMappingException(message, "Failed to create connection");
 			}
+			if (logger.isDebugEnabled()) {
+				logger.debug("Got Connection " + connection.getConnectionId());
+			}
 			connection.send(message);
 		} catch (Exception e) {
+			String connectionId = null; 
+			if (this.connection != null) {
+				connectionId = this.connection.getConnectionId();
+			}
 			this.connection = null;
 			if (e instanceof MessageMappingException) {
 				throw (MessageMappingException) e;
 			}
-			throw new MessageMappingException(message, "Failed to map message", e);
+			throw new MessageMappingException(message, "Failed to map message using " + connectionId, e);
 		}
 	}
 

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

@@ -31,6 +31,8 @@ import org.springframework.util.Assert;
  */
 public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory {
 
+	protected TcpConnection theConnection;
+
 	/**
 	 * Constructs a factory that will established connections to the host and port.
 	 * @param host The host.

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

@@ -79,6 +79,8 @@ public abstract class AbstractConnectionFactory
 
 	protected boolean active;
 
+	protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
+
 	/**
 	 * Sets socket attributes on the socket.
 	 * @param socket The socket.
@@ -270,6 +272,13 @@ public abstract class AbstractConnectionFactory
 		this.mapper = mapper;
 	}
 
+	/**
+	 * @return the singleUse
+	 */
+	public boolean isSingleUse() {
+		return singleUse;
+	}
+
 	/**
 	 * If true, sockets created by this factory will be used once.
 	 * @param singleUse
@@ -278,10 +287,15 @@ public abstract class AbstractConnectionFactory
 		this.singleUse = singleUse;
 	}
 
+	
 	public void setPoolSize(int poolSize) {
 		this.poolSize = poolSize;
 	}
 
+	public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
+		this.interceptorFactoryChain = interceptorFactoryChain;
+	}
+
 	/**
 	 * Closes the server.
 	 */
@@ -307,6 +321,23 @@ public abstract class AbstractConnectionFactory
 		this.close();
 	}
 
+	protected TcpConnection wrapConnection(TcpConnection connection) throws Exception {
+		if (this.interceptorFactoryChain == null) {
+			return connection;
+		}
+		for (TcpConnectionInterceptorFactory factory :
+				this.interceptorFactoryChain.getInterceptorFactories()) {
+			TcpConnectionInterceptor wrapper = factory.getInterceptor();
+			wrapper.setTheConnection(connection);
+			// if no ultimate listener, register each wrapper in turn
+			if (this.listener == null) {
+				connection.registerListener(wrapper);
+			}
+			connection = wrapper;
+		}
+		return connection;
+	}
+
 	
 
 }

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

@@ -65,7 +65,9 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 	 * @param socket The new socket. 
 	 */
 	protected void initializeConnection(TcpConnection connection, Socket socket) {
-		connection.registerListener(this.listener);
+		if (this.listener != null) {
+			connection.registerListener(this.listener);
+		}
 		connection.registerSender(this.sender);
 		connection.setMapper(this.mapper);
 		connection.setInputConverter(this.inputConverter);

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

@@ -36,10 +36,10 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 
 	protected Log logger = LogFactory.getLog(this.getClass());
 	
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings("rawtypes")
 	protected InputStreamingConverter inputConverter;
 	
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings("rawtypes")
 	protected OutputStreamingConverter outputConverter;
 	
 	protected TcpMessageMapper mapper;
@@ -49,7 +49,15 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 	protected TcpSender sender;
 
 	protected boolean singleUse;
+
+	protected final boolean server;
+
+	protected String connectionId;
 	
+	public AbstractTcpConnection(boolean server) {
+		this.server = server;
+	}
+
 	/**
 	 * Closes this connection.
 	 */
@@ -78,6 +86,14 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 		}
 	}
 
+	/**
+	 * 
+	 * @return the input converter
+	 */
+	public InputStreamingConverter<?> getInputConverter() {
+		return inputConverter;
+	}
+
 	/**
 	 * @param inputConverter the input converter to set
 	 */
@@ -85,6 +101,14 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 		this.inputConverter = inputConverter;
 	}
 
+	/**
+	 * 
+	 * @return the output converter
+	 */
+	public OutputStreamingConverter<?> getOutputConverter() {
+		return outputConverter;
+	}
+
 	/**
 	 * @param outputConverter the output converter to set 
 	 */
@@ -135,4 +159,8 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 		return this.singleUse;
 	}
 
+	public boolean isServer() {
+		return server;
+	}
+
 }

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

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp.connection;
+
+
+import org.springframework.commons.serializer.InputStreamingConverter;
+import org.springframework.commons.serializer.OutputStreamingConverter;
+import org.springframework.integration.Message;
+
+/**
+ * Base class for TcpConnectionIntercepters; passes all method calls through
+ * to the underlying {@link TcpConnection}.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionInterceptor {
+
+	private TcpConnection theConnection;
+	
+	private TcpListener tcpListener;
+
+	private TcpSender tcpSender;
+
+	public void close() {
+		this.theConnection.close();
+	}
+
+	public boolean isOpen() {
+		return this.theConnection.isOpen();
+	}
+
+	public Object getPayload() throws Exception {
+		return this.theConnection.getPayload();
+	}
+
+	public String getHostName() {
+		return this.theConnection.getHostName();
+	}
+
+	public String getHostAddress() {
+		return this.theConnection.getHostAddress();
+	}
+
+	public int getPort() {
+		return this.theConnection.getPort();
+	}
+
+	public void registerListener(TcpListener listener) {
+		this.theConnection.registerListener(this);
+		this.tcpListener = listener;
+	}
+
+	public void registerSender(TcpSender sender) {
+		this.tcpSender = sender;
+		this.theConnection.registerSender(this);
+	}
+
+	public String getConnectionId() {
+		return this.theConnection.getConnectionId();
+	}
+
+	public boolean isSingleUse() {
+		return this.theConnection.isSingleUse();
+	}
+
+	public void run() {
+		this.theConnection.run();
+	}
+
+	public void setSingleUse(boolean singleUse) {
+		this.theConnection.setSingleUse(singleUse);
+	}
+
+	public void setMapper(TcpMessageMapper mapper) {
+		this.theConnection.setMapper(mapper);
+	}
+
+	public InputStreamingConverter<?> getInputConverter() {
+		return this.theConnection.getInputConverter();
+	}
+
+	public void setInputConverter(InputStreamingConverter<?> inputConverter) {
+		this.theConnection.setInputConverter(inputConverter);
+	}
+
+	public OutputStreamingConverter<?> getOutputConverter() {
+		return this.theConnection.getOutputConverter();
+	}
+
+	public void setOutputConverter(OutputStreamingConverter<?> outputConverter) {
+		this.theConnection.setOutputConverter(outputConverter);
+	}
+
+	public boolean isServer() {
+		return this.theConnection.isServer();
+	}
+
+	public void onMessage(Message<?> message) {
+		if (this.tcpListener == null) {
+			throw new NoListenerException("No listener registered for message reception");
+		}
+		this.tcpListener.onMessage(message);
+	}
+
+	public void send(Message<?> message) throws Exception {
+		this.theConnection.send(message);
+	}
+
+	/**
+	 * Returns the underlying connection (or next interceptor)
+	 * @return the connection
+	 */
+	public TcpConnection getTheConnection() {
+		return this.theConnection;
+	}
+
+	/**
+	 * Sets the underlying connection (or next interceptor)
+	 * @param theConnection the connection
+	 */
+	public void setTheConnection(TcpConnection theConnection) {
+		this.theConnection = theConnection;
+	}
+
+	/**
+	 * @return the listener
+	 */
+	public TcpListener getListener() {
+		return tcpListener;
+	}
+
+	public void addNewConnection(TcpConnection connection) {
+		if (this.tcpSender != null) {
+			this.tcpSender.addNewConnection(this);
+		}
+	}
+
+	public void removeDeadConnection(TcpConnection connection) {
+		if (this.tcpSender != null) {
+			this.tcpSender.removeDeadConnection(this);
+		}
+	}
+
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp.connection;
+
+import org.springframework.core.NestedRuntimeException;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class NoListenerException extends NestedRuntimeException {
+
+	private static final long serialVersionUID = -5644042657316429223L;
+
+	public NoListenerException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+
+	public NoListenerException(String msg) {
+		super(msg);
+	}
+
+}

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

@@ -106,19 +106,42 @@ public interface TcpConnection extends Runnable {
 	 */
 	public boolean isSingleUse(); 
 	
+	/**
+	 * 
+	 * @return True if connection is used once.
+	 */
+	public boolean isServer(); 
+	
 	/**
 	 * @param mapper the mapper
 	 */
 	public void setMapper(TcpMessageMapper mapper);
 
+	/**
+	 * 
+	 * @return the input converter
+	 */
+	public InputStreamingConverter<?> getInputConverter();
+
 	/**
 	 * @param inputConverter the inputConverter to set
 	 */
 	public void setInputConverter(InputStreamingConverter<?> inputConverter);
 
+	/**
+	 * 
+	 * @return the output converter
+	 */
+	public OutputStreamingConverter<?> getOutputConverter();
+	
 	/**
 	 * @param outputConverter the outputConverter to set
 	 */
 	public void setOutputConverter(OutputStreamingConverter<?> outputConverter);
 	
+	/**
+	 * @return this connection's listener
+	 */
+	public TcpListener getListener();
+	
 }

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

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2002-2010 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;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public interface TcpConnectionInterceptor extends TcpConnection, TcpListener, TcpSender {
+	
+	public void setTheConnection(TcpConnection connection);
+
+}

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2002-2010 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;
+
+
+/**
+ * Base class for TcpConnectionInterceptorFactories. Subclasses create prototype beans by
+ * default.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public abstract class TcpConnectionInterceptorFactory {
+
+	public abstract TcpConnectionInterceptor getInterceptor();
+}
+

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2010 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;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class TcpConnectionInterceptorFactoryChain {
+
+	private TcpConnectionInterceptorFactory[] interceptorFactories;
+
+	public TcpConnectionInterceptorFactory[] getInterceptorFactories() {
+		return interceptorFactories;
+	}
+
+	public void setInterceptors(TcpConnectionInterceptorFactory[] interceptorFactories) {
+		this.interceptorFactories = interceptorFactories;
+	}
+	
+}

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

@@ -34,11 +34,4 @@ public interface TcpListener {
 	 */
 	public abstract void onMessage(Message<?> message);
 	
-	/**
-	 * Return true if the connection factory is a server
-	 * and it is listening.
-	 * @return true if listening
-	 */
-	public boolean isListening();
-
 }

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

@@ -29,8 +29,6 @@ import javax.net.SocketFactory;
 public class TcpNetClientConnectionFactory extends
 		AbstractClientConnectionFactory {
 
-	protected TcpNetConnection theConnection;
-	
 	/**
 	 * Creates a TcpNetClientConnectionFactory for connections to the host and port.
 	 * @param host the host
@@ -45,14 +43,15 @@ public class TcpNetClientConnectionFactory extends
 	 * true, a new connection is returned; otherwise a single connection is
 	 * reused for all requests while the connection remains open.
 	 */
-	public TcpNetConnection getConnection() throws Exception {
+	public TcpConnection getConnection() throws Exception {
 		if (this.theConnection != null && this.theConnection.isOpen()) {
 			return this.theConnection;
 		}
 		logger.debug("Opening new socket connection to " + this.host + ":" + this.port);
 		Socket socket = SocketFactory.getDefault().createSocket(this.host, this.port);
 		setSocketAttributes(socket);
-		TcpNetConnection connection = new TcpNetConnection(socket, false);
+		TcpConnection connection = new TcpNetConnection(socket, false);
+		connection = wrapConnection(connection);
 		initializeConnection(connection, socket);
 		this.taskExecutor.execute(connection);
 		if (!this.singleUse) {

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

@@ -35,8 +35,6 @@ public class TcpNetConnection extends AbstractTcpConnection {
 
 	private final Socket socket; 
 	
-	private final boolean server;
-	
 	/**
 	 * Constructs a TcpNetConnection for the socket.
 	 * @param socket the socket
@@ -44,8 +42,9 @@ public class TcpNetConnection extends AbstractTcpConnection {
 	 * a result of an incoming request.
 	 */
 	public TcpNetConnection(Socket socket, boolean server) {
+		super(server);
 		this.socket = socket;
-		this.server = server;
+		getConnectionId();
 	}
 	
 	/**
@@ -121,15 +120,6 @@ public class TcpNetConnection extends AbstractTcpConnection {
 				}
 				break;
 			}
-			/*
-			 * For single use sockets, we close after receipt if we are on the client
-			 * side, or the server side has no outbound adapter registered
-			 */
-			if (this.singleUse && (!this.server || this.sender == null)) {
-				logger.debug("Closing single use socket after inbound message");
-				this.close();
-				okToRun = false;
-			}
 			if (logger.isDebugEnabled())
 				logger.debug("Message received " + message);
 			try {
@@ -139,13 +129,35 @@ public class TcpNetConnection extends AbstractTcpConnection {
 				}
 				listener.onMessage(message);
 			} catch (Exception e) {
-				logger.error("Exception sending meeeage: " + message, e);				
+				if (e instanceof NoListenerException) {
+					if (this.singleUse) {
+						logger.debug("Closing single use socket after inbound message " + this.connectionId);
+						this.close();
+						okToRun = false;
+					} else {
+						logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
+					}
+				} else {
+					logger.error("Exception sending meeeage: " + message, e);				
+				}
+			}
+			/*
+			 * For single use sockets, we close after receipt if we are on the client
+			 * side, or the server side has no outbound adapter registered
+			 */
+			if (this.singleUse && this.server && this.sender == null) {
+				logger.debug("Closing single use socket after inbound message " + this.connectionId);
+				this.close();
+				okToRun = false;
 			}
 		}
 	}
 
 	public String getConnectionId() {
-		return SocketIoUtils.getSocketId(this.socket);
+		if (this.connectionId == null) {
+			this.connectionId = SocketIoUtils.getSocketId(this.socket);
+		}
+		return this.connectionId;
 	}
 
 

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

@@ -69,11 +69,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 				final Socket socket = serverSocket.accept();
 				logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
 				setSocketAttributes(socket);
-				TcpNetConnection connection = new TcpNetConnection(socket, true);
+				TcpConnection connection = new TcpNetConnection(socket, true);
+				connection = wrapConnection(connection);
 				this.initializeConnection(connection, socket);
 				this.taskExecutor.execute(connection);
 			}
-		} catch (IOException e) {
+		} catch (Exception e) {
 			this.listening = false;
 			if (this.active) {
 				logger.error("Error on ServerSocket", e);

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

@@ -39,8 +39,6 @@ import java.util.concurrent.LinkedBlockingQueue;
 public class TcpNioClientConnectionFactory extends
 		AbstractClientConnectionFactory {
 
-	protected TcpNioConnection theConnection;
-	
 	protected boolean usingDirectBuffers;
 	
 	private Selector selector;
@@ -63,7 +61,7 @@ public class TcpNioClientConnectionFactory extends
 	 * true, a new connection is returned; otherwise a single connection is
 	 * reused for all requests while the connection remains open.
 	 */
-	public TcpNioConnection getConnection() throws Exception {
+	public TcpConnection getConnection() throws Exception {
 		int n = 0;
 		while (this.selector == null) {
 			Thread.sleep(100);
@@ -78,13 +76,14 @@ public class TcpNioClientConnectionFactory extends
 		SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.host, this.port));
 		setSocketAttributes(socketChannel.socket());
 		TcpNioConnection connection = new TcpNioConnection(socketChannel, false);
+		connection.setUsingDirectBuffers(this.usingDirectBuffers);
 		if (this.taskExecutor == null) {
 			connection.setTaskExecutor(Executors.newSingleThreadExecutor());
 		} else {
 			connection.setTaskExecutor(this.taskExecutor);
 		}
-		initializeConnection(connection, socketChannel.socket());
-		connection.setUsingDirectBuffers(this.usingDirectBuffers);
+		TcpConnection wrappedConnection = wrapConnection(connection);
+		initializeConnection(wrappedConnection, socketChannel.socket());
 		socketChannel.configureBlocking(false);
 		if (this.soTimeout > 0) {
 			connection.setLastRead(System.currentTimeMillis());
@@ -93,9 +92,9 @@ public class TcpNioClientConnectionFactory extends
 		newChannels.add(socketChannel);
 		selector.wakeup();
 		if (!this.singleUse) {
-			this.theConnection = connection;
+			this.theConnection = wrappedConnection;
 		}
-		return connection;
+		return wrappedConnection;
 	}
 
 	/**

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

@@ -43,8 +43,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private final SocketChannel socketChannel;
 	
-	private final boolean server;
-	
 	private OutputStream channelOutputStream;
 	
 	private PipedOutputStream pipedOutputStream;
@@ -62,7 +60,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	private boolean active = true;
 	
 	private long lastRead;
-	
+
 	/**
 	 * Constructs a TcpNetConnection for the SocketChannel.
 	 * @param socketChannel the socketChannel
@@ -70,11 +68,15 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	 * a result of an incoming request.
 	 */
 	public TcpNioConnection(SocketChannel socketChannel, boolean server) throws Exception {
+		super(server);
 		this.socketChannel = socketChannel;
-		this.server = server;
 		this.pipedInputStream = new PipedInputStream();
 		this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream);
 		this.channelOutputStream = new ChannelOutputStream();
+		getConnectionId();
+		if (this.connectionId == null) {
+			throw new Exception("Null id");
+		}
 	}
 	
 	public void close() {
@@ -166,7 +168,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private synchronized void convertAndSend() throws IOException {
 		if (this.pipedInputStream.available() <= 0) {
-			System.err.println("NO WORK");
 			return;
 		}
 		Message<?> message = null;
@@ -175,7 +176,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		} catch (Exception e) {
 			this.close();
 			if (e instanceof SocketTimeoutException && this.singleUse) {
-				logger.debug("Closing single use socket after timeout");				
+				if (logger.isDebugEnabled()) {
+					logger.debug("Closing single use socket after timeout " + this.connectionId);				
+				}
 			} else {
 				if (!(e instanceof SoftEndOfStreamException)) {
 					logger.error("Read exception " +
@@ -187,20 +190,30 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			}
 			return;
 		}			
-		/*
-		 * For single use sockets, we close after receipt if we are on the client
-		 * side, or the server side has no outbound adapter registered
-		 */
-		if (this.singleUse && (!this.server || this.sender == null)) {
-			logger.debug("Closing single use socket after inbound message");
-			this.close();
-		}
+
 		try {
 			if (message != null) {
 				listener.onMessage(message);
 			}
 		} catch (Exception e) {
-			logger.error("Exception sending meeeage: " + message, e);
+			if (e instanceof NoListenerException) {
+				if (this.singleUse) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Closing single use channel after inbound message " + this.connectionId);
+					}
+					this.close();
+				}
+			} else {
+				logger.error("Exception sending meeeage: " + message, e);
+			}
+		}
+		/*
+		 * For single use sockets, we close after receipt if we are on the client
+		 * side, or the server side has no outbound adapter registered
+		 */
+		if (this.singleUse && this.server && this.sender == null) {
+			logger.debug("Closing single use cbannel after inbound message " + this.connectionId);
+			this.close();
 		}
 	}
 	
@@ -268,7 +281,10 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	public String getConnectionId() {
-		return SocketIoUtils.getSocketId(this.socketChannel.socket());
+		if (this.connectionId == null) {
+			this.connectionId = SocketIoUtils.getSocketId(this.socketChannel.socket());			
+		}
+		return this.connectionId;
 	}
 
 	/**
@@ -326,7 +342,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			doWrite(buffer);
 		}
 		
-		private void doWrite(ByteBuffer buffer) throws IOException {
+		private synchronized void doWrite(ByteBuffer buffer) throws IOException {
 			socketChannel.write(buffer);
 			int remaining = buffer.remaining();
 			if (remaining == 0) {

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

@@ -187,8 +187,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {
 		try {
 			TcpNioConnection connection = new TcpNioConnection(socketChannel, true);
-			this.initializeConnection(connection, socketChannel.socket());
 			connection.setUsingDirectBuffers(this.usingDirectBuffers);
+			TcpConnection wrappedConnection = wrapConnection(connection);
+			this.initializeConnection(wrappedConnection, socketChannel.socket());
 			return connection;
 		} catch (Exception e) {
 			logger.error("Failed to establish new incoming connection", e);

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

@@ -372,6 +372,15 @@ its configuration specifies the number of threads.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="interceptor-factory-chain" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.ip.tcp.connection.InterceptorFactoryChain"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>