INT-1145 INT-1146 Add 'close' attribute to tcp gateways and inbound adapters

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

@@ -42,6 +42,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
 
 	protected volatile boolean active;
 
+	protected volatile boolean listening;
+
 
 	public AbstractInternetProtocolReceivingChannelAdapter(int port) {
 		this.port = port;
@@ -95,4 +97,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
 		this.active = false;
 	}
 
+	public boolean isListening() {
+		return listening;
+	}
+
 }

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

@@ -87,6 +87,8 @@ public abstract class IpAdapterParserUtils {
 
 	static final String SO_TRAFFIC_CLASS = "so-traffic-class";
 
+	static final String CLOSE = "close";
+
 
 	/**
 	 * Adds a constructor-arg to the provided bean definition builder 
@@ -172,6 +174,22 @@ public abstract class IpAdapterParserUtils {
 		return multicast;
 	}
 
+	/**
+	 * Sets the close attribute, if present.
+	 * @param element
+	 */
+	static void setClose(Element element, BeanDefinitionBuilder builder) {
+		String close = element.getAttribute(IpAdapterParserUtils.CLOSE);
+		if (!StringUtils.hasText(close)) {
+			close = "false";
+		}
+		if (close.equals("true")) {
+			builder.addPropertyValue(
+					Conventions.attributeNameToPropertyName(IpAdapterParserUtils.CLOSE),
+					close);
+		}
+	}
+
 	/**
 	 * Gets the use-nio attribute, if present; if not returns 'false'.
 	 * @param element

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

@@ -124,6 +124,7 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
 				IpAdapterParserUtils.USING_DIRECT_BUFFERS);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
 				IpAdapterParserUtils.SO_KEEP_ALIVE);
+		IpAdapterParserUtils.setClose(element, builder);
 		return builder;
 	}
 

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

@@ -26,6 +26,7 @@ import org.w3c.dom.Element;
  * Parser for the <outbound-gateway> element of the integration 'jms' namespace.
  *
  * @author Gary Russell
+ * @since 2.0
  */
 public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 
@@ -44,6 +45,7 @@ public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
 				IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME); 
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
+		IpAdapterParserUtils.setClose(element, builder);
 		return builder;
 	}
 

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

@@ -33,6 +33,7 @@ import java.io.IOException;
  * when the format is {@link MessageFormats#FORMAT_CUSTOM}. 
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public abstract class AbstractSocketReader implements SocketReader, MessageFormats {
@@ -52,21 +53,21 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma
 	 * @return True when a message is completely assembled.
 	 * @throws IOException
 	 */
-	protected abstract boolean assembleDataLengthFormat() throws IOException;
+	protected abstract int assembleDataLengthFormat() throws IOException;
 	
 	/**
 	 * Assembles data in format {@link #FORMAT_STX_ETX}.
 	 * @return True when a message is completely assembled.
 	 * @throws IOException
 	 */
-	protected abstract boolean assembleDataStxEtxFormat() throws IOException;
+	protected abstract int assembleDataStxEtxFormat() throws IOException;
 	
 	/**
 	 * Assembles data in format {@link #FORMAT_CRLF}.
 	 * @return True when a message is completely assembled.
 	 * @throws IOException
 	 */
-	protected abstract boolean assembleDataCrLfFormat() throws IOException;
+	protected abstract int assembleDataCrLfFormat() throws IOException;
 	
 	/**
 	 * Assembles data in format {@link #FORMAT_CUSTOM}. Implementations must
@@ -76,9 +77,9 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma
 	 * @return True when a message is completely assembled.
 	 * @throws IOException
 	 */
-	protected abstract boolean assembleDataCustomFormat() throws IOException;
+	protected abstract int assembleDataCustomFormat() throws IOException;
 	
-	public boolean assembleData() throws IOException {
+	public int assembleData() throws IOException {
 		try {
 			switch (this.messageFormat) {
 			case FORMAT_LENGTH_HEADER:

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

@@ -17,6 +17,9 @@ package org.springframework.integration.ip.tcp;
 
 import java.io.IOException;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 /**
  * Abstract SocketWriter that handles data in 3 standard, and one custom
  * format. The default format is {@link MessageFormats#FORMAT_LENGTH_HEADER} in which 
@@ -31,14 +34,18 @@ import java.io.IOException;
  * appropriate implementation, and provide an implementation for 
  * {@link #writeCustomFormat(byte[])} which is invoked by {@link #write(byte[])}
  * when the format is {@link MessageFormats#FORMAT_CUSTOM}. 
+ * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public abstract class AbstractSocketWriter implements SocketWriter, MessageFormats {
 
 	protected int messageFormat = FORMAT_LENGTH_HEADER;
 
-	/*
+    protected final Log logger = LogFactory.getLog(this.getClass());
+
+    /*
 	 * @see org.springframework.integration.ip.tcp.SocketWriter#write(byte[])
 	 */
 	public synchronized void write(byte[] bytes) throws IOException {

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

@@ -29,6 +29,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  * are provided.
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public abstract class AbstractTcpReceivingChannelAdapter extends
@@ -42,7 +43,9 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
 
 	protected volatile boolean soKeepAlive;
 
-	protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
+	protected volatile int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
+	
+	protected volatile boolean close;
 	
 	/**
 	 * Constructs a receiving channel adapter that listens on the port.
@@ -122,4 +125,11 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
 		this.poolSize = poolSize;
 	}
 
+	/**
+	 * @param close the close to set
+	 */
+	public void setClose(boolean close) {
+		this.close = close;
+	}
+
 }

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

@@ -32,6 +32,7 @@ import org.springframework.integration.message.MessageMappingException;
  * is completely assembled.
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class NetSocketReader extends AbstractSocketReader {
@@ -52,9 +53,11 @@ public class NetSocketReader extends AbstractSocketReader {
 	 * @see org.springframework.integration.ip.tcp.SocketReader#read(java.nio.ByteBuffer)
 	 */
 	@Override
-	protected boolean assembleDataLengthFormat() throws IOException {
+	protected int assembleDataLengthFormat() throws IOException {
 		byte[] lengthPart = new byte[4];
-		read(lengthPart);
+		int status = read(lengthPart, true);
+		if (status < 0)
+			return status;
 		int messageLength = ByteBuffer.wrap(lengthPart).getInt();
 		if (logger.isDebugEnabled()) {
 			logger.debug("Message length is " + messageLength);
@@ -64,27 +67,27 @@ public class NetSocketReader extends AbstractSocketReader {
 					" exceeds max message length: " + this.maxMessageSize);
 		}
 		byte[] messagePart = new byte[messageLength];
-		read(messagePart);
+		read(messagePart, false);
 		assembledData = messagePart;
-		return true;
+		return MESSAGE_COMPLETE;
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat()
 	 */
 	@Override
-	protected boolean assembleDataStxEtxFormat() throws IOException {
+	protected int assembleDataStxEtxFormat() throws IOException {
 		InputStream inputStream = socket.getInputStream();
-		if (inputStream.read() != STX)
+		int bite = inputStream.read();
+		if (bite < 0) {
+			return bite;
+		}
+		if (bite != STX)
 			throw new MessageMappingException("Expected STX to begin message");
 		byte[] buffer = new byte[this.maxMessageSize];
 		int n = 0;
-		int bite;
 		while ((bite = inputStream.read()) != ETX) {
-			if (bite < 0) {
+			checkClosure(bite);
-				logger.debug("Socket closed");				
-				throw new IOException("Socket Closed");
-			}
 			buffer[n++] = (byte) bite;
 			if (n >= this.maxMessageSize) {
 				throw new IOException("ETX not found before max message length: "
@@ -93,24 +96,31 @@ public class NetSocketReader extends AbstractSocketReader {
 		}
 		assembledData = new byte[n];
 		System.arraycopy(buffer, 0, assembledData, 0, n);
-		return true;
+		return MESSAGE_COMPLETE;
+	}
+
+	private void checkClosure(int bite) throws IOException {
+		if (bite < 0) {
+			logger.debug("Socket closed");				
+			throw new IOException("Socket closed");
+		}
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat()
 	 */
 	@Override
-	protected boolean assembleDataCrLfFormat() throws IOException {
+	protected int assembleDataCrLfFormat() throws IOException {
 		InputStream inputStream = socket.getInputStream();
 		byte[] buffer = new byte[this.maxMessageSize];
 		int n = 0;
 		int bite;
 		while (true) {
 			bite = inputStream.read();
-			if (bite < 0) {
+			if (bite < 0 && n == 0) {
-				logger.debug("Socket closed");				
+				return bite;
-				throw new IOException("Socket Closed");
 			}
+			checkClosure(bite);
 			if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
 				break;
 			buffer[n++] = (byte) bite;
@@ -121,7 +131,7 @@ public class NetSocketReader extends AbstractSocketReader {
 		};
 		assembledData = new byte[n-1];
 		System.arraycopy(buffer, 0, assembledData, 0, n-1);
-		return true;
+		return MESSAGE_COMPLETE;
 	}
 
 	/**
@@ -132,7 +142,7 @@ public class NetSocketReader extends AbstractSocketReader {
 	 * 
 	 */
 	@Override
-	protected boolean assembleDataCustomFormat() throws IOException {
+	protected int assembleDataCustomFormat() throws IOException {
 		throw new UnsupportedOperationException("Need to subclass for this format");
 	}
 
@@ -149,19 +159,23 @@ public class NetSocketReader extends AbstractSocketReader {
 	 * Reads data from the socket and puts the data in buffer. Blocks until
 	 * buffer is full or a socket timeout occurs.
 	 * @param buffer
+	 * @param header true if we are reading the header
+	 * @return < 0 if socket closed and not in the middle of a message
 	 * @throws IOException
 	 */
-	protected void read(byte[] buffer) throws IOException {
+	protected int read(byte[] buffer, boolean header) throws IOException {
 		int lengthRead = 0;
 		int needed = buffer.length;
 		while (lengthRead < needed) {
 			int len;
 			len = socket.getInputStream().read(buffer, lengthRead,
 					needed - lengthRead);
-			if (len < 0) {
+			if (len < 0 && header && lengthRead == 0) {
-				logger.debug("Socket closed");				
+				return len;
-				throw new IOException("Socket Closed");
 			}
+			if (len < 0)
+				logger.debug("socket closed after " + lengthRead + " of " + needed);
+			checkClosure(len);
 			lengthRead += len;
 			if (logger.isDebugEnabled()) {
 				logger.debug("Read " + len + " bytes, buffer is now at " + 
@@ -169,7 +183,7 @@ public class NetSocketReader extends AbstractSocketReader {
 							 needed);
 			}
 		}
-		
+		return 0;
 	}
 	
 	/* (non-Javadoc)

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

@@ -25,6 +25,7 @@ import java.nio.ByteBuffer;
  * data is wrapped in a wire protocol based on the messageFormat property.
  * 
  * @author Gary Russell
+ * @since 2.0
  */
 public class NetSocketWriter extends AbstractSocketWriter {
 
@@ -87,6 +88,8 @@ public class NetSocketWriter extends AbstractSocketWriter {
 	protected void doClose() {
 		try {
 			socket.close();
-		} catch (IOException e) {}
+		} catch (IOException e) {
+			logger.error("Error on close", e);
+		}
 	}
 }

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

@@ -30,6 +30,7 @@ import org.springframework.integration.message.MessageMappingException;
  * A non-blocking SocketReader that reads from a {@link java.nio.channels.SocketChannel}.
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class NioSocketReader extends AbstractSocketReader {
@@ -71,13 +72,13 @@ public class NioSocketReader extends AbstractSocketReader {
 	 * @see org.springframework.integration.ip.tcp.SocketReader#assembleData()
 	 */
 	@Override
-	public boolean assembleDataLengthFormat() throws IOException {
+	public int assembleDataLengthFormat() throws IOException {
 		if (lengthPart == null) {
 			lengthPart = allocate(4);
 		}
 		if (lengthPart.hasRemaining()) {
 			readChannel(lengthPart);
-			return false;
+			return MESSAGE_INCOMPLETE;
 		}
 		if (dataPart == null) {
 			lengthPart.flip();
@@ -94,19 +95,19 @@ public class NioSocketReader extends AbstractSocketReader {
 		if (dataPart.hasRemaining()) {
 			readChannel(dataPart);
 			if (dataPart.hasRemaining()) {
-				return false;
+				return MESSAGE_INCOMPLETE;
 			}
 		} 
 		assembledData = dataPart.array();
 		lengthPart = dataPart = null;
-		return true;
+		return MESSAGE_COMPLETE;
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat()
 	 */
 	@Override
-	protected boolean assembleDataStxEtxFormat() throws IOException {
+	protected int assembleDataStxEtxFormat() throws IOException {
 		if (readChannelNonDeterministic()) {
 			byte bite = rawBuffer.get();
 			int count = 0;
@@ -120,12 +121,12 @@ public class NioSocketReader extends AbstractSocketReader {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Incomplete message, consumed 1 byte");
 					}
-					return false;
+					return MESSAGE_INCOMPLETE;
 				}
 			} else {
 				if (bite == ETX) {
 					finishAssembly();
-					return true;
+					return MESSAGE_COMPLETE;
 				}
 				buildBuffer.put(bite);
 				count++;
@@ -139,7 +140,7 @@ public class NioSocketReader extends AbstractSocketReader {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Incomplete message, consumed " + count + " bytes");
 					}
-					return false;
+					return MESSAGE_INCOMPLETE;
 				}
 				bite = rawBuffer.get();
 				if (bite == ETX) {
@@ -156,13 +157,13 @@ public class NioSocketReader extends AbstractSocketReader {
 				logger.debug("Consumed " + count + " bytes");
 			}
 			finishAssembly();
-			return true;
+			return MESSAGE_COMPLETE;
 		} else {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Incomplete message, consumed 0 bytes");
 			}
 		}
-		return false;
+		return MESSAGE_INCOMPLETE;
 	}
 
 	/**
@@ -180,7 +181,7 @@ public class NioSocketReader extends AbstractSocketReader {
 	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat()
 	 */
 	@Override
-	protected boolean assembleDataCrLfFormat() throws IOException {
+	protected int assembleDataCrLfFormat() throws IOException {
 		if (readChannelNonDeterministic()) {
 			int count = 0;
 			while (true) {
@@ -188,7 +189,7 @@ public class NioSocketReader extends AbstractSocketReader {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Incomplete message, consumed " + count + " bytes");
 					}
-					return false;
+					return MESSAGE_INCOMPLETE;
 				}
 				byte bite = rawBuffer.get();
 				if (bite == '\n' && buildBuffer.position() > 0) {
@@ -209,13 +210,13 @@ public class NioSocketReader extends AbstractSocketReader {
 				logger.debug("Consumed " + count + " bytes");
 			}
 			finishAssembly();
-			return true;
+			return MESSAGE_COMPLETE;
 		} else {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Incomplete message, consumed 0 bytes");
 			}
 		}
-		return false;
+		return MESSAGE_INCOMPLETE;
 	}
 
 	/**
@@ -226,7 +227,7 @@ public class NioSocketReader extends AbstractSocketReader {
 	 * 
 	 */
 	@Override
-	protected boolean assembleDataCustomFormat() throws IOException {
+	protected int assembleDataCustomFormat() throws IOException {
 		throw new UnsupportedOperationException("Need to subclass for this format");
 	}
 

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

@@ -27,6 +27,7 @@ import java.util.concurrent.LinkedBlockingQueue;
  * data is wrapped in a wire protocol based on the messageFormat property.
  *
  * @author Gary Russell
+ * @since 2.0
  */
 public class NioSocketWriter extends AbstractSocketWriter {
 
@@ -250,7 +251,9 @@ public class NioSocketWriter extends AbstractSocketWriter {
 	protected void doClose() {
 		try {
 			channel.close();
-		} catch (IOException e) {}
+		} catch (IOException e) {
+			logger.error("Error on close", e);
+		}
 	}
 
 }

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.integration.ip.tcp;
 
+import java.io.IOException;
 import java.net.Socket;
 import java.net.SocketException;
 
@@ -34,6 +35,7 @@ import org.springframework.integration.message.MessageMappingException;
  * number of concurrent connections expected.
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
@@ -62,6 +64,8 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
 	
 	protected Class<NetSocketWriter> customSocketWriterClass;
 	
+	protected boolean close;
+	
 	@Override
 	protected void doStart() {
 		super.doStart();
@@ -86,6 +90,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
 		this.delegate.setSoTimeout(soTimeout);
 		this.delegate.setTaskScheduler(getTaskScheduler());
 		this.delegate.setCustomSocketReaderClassName(customSocketReaderClassName);
+		this.delegate.setClose(close);
 		super.onInit();
 	}
 
@@ -189,6 +194,17 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
 		}
 	}
 
+	/**
+	 * @param close the close to set
+	 */
+	public void setClose(boolean close) {
+		this.close = close;
+	}
+	
+	public boolean isListening() {
+		return delegate.isListening();
+	}
+
 	private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter {
 
 		/**
@@ -206,6 +222,13 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
 					customSocketWriterClass, socket);
 			try {
 				writer.write(this.mapper.fromMessage(message));
+				if (close) {
+					try {
+						socket.close();
+					} catch (IOException ioe) {
+						logger.error("Error on close", ioe);
+					}
+				}
 			} catch (Exception e) {
 				throw new MessageMappingException("Failed to map and send response", e);
 			}

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

@@ -50,6 +50,8 @@ public class SimpleTcpNetOutboundGateway extends
 	
 	protected NetSocketReader reader;
 	
+	protected boolean close;
+	
 	/**
 	 * Constructs a SimpleTcpNetOutboundGateway that sends data to the 
 	 * specified host and port, and waits for a response.
@@ -67,9 +69,9 @@ public class SimpleTcpNetOutboundGateway extends
 	@Override
 	protected synchronized Object handleRequestMessage(Message<?> requestMessage) {
 		this.handler.handleMessage(requestMessage);
+		Socket socket = this.handler.getSocket();
 		if (this.reader == null ||
-			this.reader.getSocket() != handler.getSocket()) {
+			this.reader.getSocket() != socket) { // might have re-opened on error
-			Socket socket = this.handler.getSocket();
 			this.reader = SocketIoUtils.createNetReader(this.messageFormat,
 				this.customSocketReaderClass, socket, this.maxMessageSize,
 				this.soReceiveBufferSize);
@@ -77,6 +79,10 @@ public class SimpleTcpNetOutboundGateway extends
 		try {
 			this.reader.assembleData();  // Net... always returns true
 			byte[] bytes = this.reader.getAssembledData();
+			if (close) {
+				logger.debug("Closing socket because close=true");
+				this.handler.close();
+			}
 			return bytes;
 		} catch (Exception e) {
 			this.reader = null;
@@ -207,4 +213,11 @@ public class SimpleTcpNetOutboundGateway extends
 		this.setOutputChannel(replyChannel);
 	}
 
+	/**
+	 * @param close the close to set
+	 */
+	public void setClose(boolean close) {
+		this.close = close;
+	}
+
 }

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

@@ -23,19 +23,26 @@ import java.net.Socket;
  * General interface for assembling message data from a TCP/IP Socket.
  * Implementations for {@link java.net.Socket} and {@link java.nio.channels.SocketChannel}
  * are provided.
+ * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public interface SocketReader {
 
+	public static int MESSAGE_INCOMPLETE = 0;
+	
+	public static int MESSAGE_COMPLETE = 1;
+	
 	/**
 	 * Reads the data the socket and assembles 
 	 * packets of data into a complete message, depending on the format of that
 	 * data.
-	 * @return true when the message is assembled.
+	 * @return MESSAGE_COMPLETE when message is assembled, otherwise MESSAGE_IMCOMPLETE, or
+	 * < 0 if socket closed before any data for a message is received.
 	 * @throws IOException 
 	 */
-	public boolean assembleData() throws IOException;
+	public int assembleData() throws IOException;
 	
 	/**
 	 * Retrieves the assembled tcp data or null if the data is not

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

@@ -30,6 +30,7 @@ import org.springframework.integration.ip.util.SocketIoUtils;
  * accordingly.
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class TcpNetReceivingChannelAdapter extends
@@ -58,6 +59,7 @@ public class TcpNetReceivingChannelAdapter extends
 			try {
 				serverSocket = ServerSocketFactory.getDefault()
 						.createServerSocket(port, Math.abs(poolSize));
+				listening = true;
 				while (true) {
 					final Socket socket = serverSocket.accept();
 					setSocketOptions(socket);
@@ -72,6 +74,7 @@ public class TcpNetReceivingChannelAdapter extends
 						serverSocket.close();
 					} catch (IOException e1) {}
 				}
+				listening = false;
 				serverSocket = null;
 				if (active) {
 					logger.error("Error on ServerSocket", e);
@@ -94,8 +97,21 @@ public class TcpNetReceivingChannelAdapter extends
 				this.soReceiveBufferSize);
 		while (true) {
 			try {
-				if (reader.assembleData()) {
+				int messageStatus = reader.assembleData();
+				if (messageStatus < 0) {
+					return;
+				}
+				if (messageStatus == SocketReader.MESSAGE_COMPLETE) {
 					processMessage(reader);
+					if (close) {
+						logger.debug("Closing socket because close=true");
+						try {
+							reader.getSocket().close();
+						} catch (IOException ioe) {
+							logger.error("Error on close", ioe);
+						}
+						break;
+					}
 				}
 			} catch (Exception e) {
 				logger.error("processMessage failed", e);

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

@@ -25,6 +25,7 @@ import org.springframework.integration.ip.util.SocketIoUtils;
 /**
  * TCP Sending Channel Adapter that that uses a {@link java.net.Socket}.
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class TcpNetSendingMessageHandler extends
@@ -57,6 +58,7 @@ public class TcpNetSendingMessageHandler extends
 	protected synchronized SocketWriter getWriter() {
 		if (this.writer == null) {
 			try {
+				logger.debug("Opening new socket connection");
 				this.socket = SocketFactory.getDefault().createSocket(this.host, this.port);
 				this.setSocketAttributes(socket);
 				NetSocketWriter writer = SocketIoUtils.createNetWriter(messageFormat,
@@ -85,4 +87,12 @@ public class TcpNetSendingMessageHandler extends
 		}
 	}
 
+	/**
+	 * Close the underlying socket and prepare to establish a new socket on
+	 * the next write.
+	 */
+	protected void close() {
+		this.writer.doClose();
+		this.writer = null;
+	}
 }

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

@@ -38,6 +38,7 @@ import org.springframework.integration.ip.util.SocketIoUtils;
  * number of threads is controlled by the poolSize property.
  * 
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class TcpNioReceivingChannelAdapter extends
@@ -65,6 +66,7 @@ public class TcpNioReceivingChannelAdapter extends
 	protected void server() {
 		try {
 			serverChannel = ServerSocketChannel.open();
+			listening = true;
 			serverChannel.configureBlocking(false);
 			serverChannel.socket().bind(new InetSocketAddress(port),
 					Math.abs(poolSize));
@@ -76,6 +78,7 @@ public class TcpNioReceivingChannelAdapter extends
 			try {
 				serverChannel.close();
 			} catch (IOException e1) { }
+			listening = false;
 			serverChannel = null;
 			if (active) {
 				logger.error("Error on ServerSocketChannel", e);
@@ -169,11 +172,23 @@ public class TcpNioReceivingChannelAdapter extends
 	private void doRead(SelectionKey key) {
 		NioSocketReader reader = (NioSocketReader) key.attachment();
 		try {
-			if (reader.assembleData()) {
+			int messageStatus = reader.assembleData();
+			if (messageStatus < 0) {
+				return;
+			}
+			if (messageStatus == SocketReader.MESSAGE_COMPLETE) {
 				Message<byte[]> message;
 					message = mapper.toMessage(reader);
 					if (message != null) {
 						sendMessage(message);
+						if (close) {
+							logger.debug("Closing channel because close=true");
+							try {
+								key.channel().close();
+							} catch (IOException ioe) {
+								logger.error("Error on close", ioe);
+							}
+						}
 					}
 			}
 		} catch (Exception e) {}

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

@@ -21,7 +21,10 @@ import org.springframework.integration.ip.util.SocketIoUtils;
 
 
 /**
+ * TCP Sending Channel Adapter that that uses a {@link java.nio.channels.SocketChannel}.
+ *  
  * @author Gary Russell
+ * @since 2.0
  *
  */
 public class TcpNioSendingMessageHandler extends
@@ -49,6 +52,7 @@ public class TcpNioSendingMessageHandler extends
 	protected synchronized SocketWriter getWriter() {
 		if (this.socketChannel == null) {
 			try {
+				logger.debug("Creating new SocketChannel");
 				this.socketChannel = SocketChannel.open(this.destinationAddress);
 				this.setSocketAttributes(socketChannel.socket());
 				NioSocketWriter writer = SocketIoUtils.createNioWriter(messageFormat, 

spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java

@@ -102,6 +102,8 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
 			this.threadPoolTaskScheduler.initialize();
 		}
 
+		listening = true;
+		
 		// Do as little as possible here so we can loop around and catch the next packet.
 		// Just schedule the packet for processing.
 		while (this.active) {
@@ -121,6 +123,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
 				throw new MessagingException("failed to receive DatagramPacket", e);
 			}
 		}
+		listening = false;
 	}
 
 	protected void sendAck(Message<byte[]> message) {

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

@@ -39,6 +39,15 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>					
+					<xsd:attribute name="close" type="xsd:string" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+If true, the socket will be closed after a message is received and sent to the
+outbound channel. If false, the socket will remain open ready to receive the 
+next message.							
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -202,6 +211,18 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
 				<xsd:attribute name="reply-timeout" type="xsd:string"/>
 				<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
 				<xsd:attribute name="receive-buffer-size" type="xsd:string" />
+					<xsd:attribute name="close" type="xsd:string" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+If true, for an outbound gateway, the socket will be closed after a response 
+is received and sent to the
+reply channel. If false, the socket will remain open and be used to send the 
+next message. If true, for an inbound gateway, the socket will be closed after
+the response is sent. If false, the socket will remain open and be used to 
+receive the next message.					
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

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

@@ -74,6 +74,7 @@
 		so-timeout="32"
 		using-direct-buffers="true"
 		using-nio="true"
+		close="true"
 	/>
 
 	<ip:inbound-channel-adapter id="testInTcpNet"
@@ -197,13 +198,45 @@
 		custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"
 		message-format="crlf"
 		host="localhost"
-		port="#{tcpIpUtils.findAvailableServerSocket(6500)}"
+		port="#{tcpIpUtils.findAvailableServerSocket(6600)}"
 		receive-buffer-size="223"
 		so-keep-alive="true"
 		so-receive-buffer-size="224"
 		so-send-buffer-size="225"
 		so-timeout="226"
-		
+		close="false"
+		/>		
+
+	<ip:inbound-gateway id="simpleInGatewayClose"
+		request-channel="tcpChannel"
+		reply-channel="replyChannel"
+		custom-socket-reader-class-name="org.springframework.integration.ip.tcp.CustomNetSocketReader"
+		custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"
+		message-format="crlf"
+		pool-size="23"
+		port="#{tcpIpUtils.findAvailableServerSocket(6700)}"
+		receive-buffer-size="123"
+		so-keep-alive="true"
+		so-receive-buffer-size="124"
+		so-send-buffer-size="125"
+		so-timeout="126"
+		close="true"
+		/>	
+
+	<ip:outbound-gateway id="simpleOutGatewayClose"
+		request-channel="tcpChannel"
+		reply-channel="replyChannel"
+		custom-socket-reader-class-name="org.springframework.integration.ip.tcp.CustomNetSocketReader"
+		custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"
+		message-format="crlf"
+		host="localhost"
+		port="#{tcpIpUtils.findAvailableServerSocket(6800)}"
+		receive-buffer-size="223"
+		so-keep-alive="true"
+		so-receive-buffer-size="224"
+		so-send-buffer-size="225"
+		so-timeout="226"
+		close="true"
 		/>		
 	
 </beans>

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

@@ -95,12 +95,21 @@ public class ParserUnitTests {
 	TcpNetSendingMessageHandler tcpOutNet;
 
 	@Autowired
+	@Qualifier(value="simpleInGateway")
 	SimpleTcpNetInboundGateway simpleTcpNetInboundGateway;
 
 	@Autowired
 	@Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#0")
 	SimpleTcpNetOutboundGateway simpleTcpNetOutboundGateway;
 	
+	@Autowired
+	@Qualifier(value="simpleInGatewayClose")
+	SimpleTcpNetInboundGateway simpleTcpNetInboundGatewayClose;
+
+	@Autowired
+	@Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#1")
+	SimpleTcpNetOutboundGateway simpleTcpNetOutboundGatewayClose;
+
 	@Test
 	public void testInUdp() {
 		DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn);
@@ -136,6 +145,7 @@ public class ParserUnitTests {
 		assertEquals(29, dfa.getPropertyValue("receiveBufferSize"));
 		assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize"));
 		assertEquals(32, dfa.getPropertyValue("soTimeout"));
+		assertEquals(false, dfa.getPropertyValue("close"));
 	}
 	
 	@Test
@@ -150,6 +160,7 @@ public class ParserUnitTests {
 		assertEquals(29, dfa.getPropertyValue("receiveBufferSize"));
 		assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize"));
 		assertEquals(32, dfa.getPropertyValue("soTimeout"));
+		assertEquals(true, dfa.getPropertyValue("close"));
 	}
 
 	@Test
@@ -163,6 +174,7 @@ public class ParserUnitTests {
 		assertEquals(29, dfa.getPropertyValue("receiveBufferSize"));
 		assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize"));
 		assertEquals(32, dfa.getPropertyValue("soTimeout"));
+		assertEquals(false, dfa.getPropertyValue("close"));
 	}
 	
 	@Test
@@ -217,7 +229,6 @@ public class ParserUnitTests {
 		assertEquals(53, dfa.getPropertyValue("soSendBufferSize"));
 		assertEquals(54, dfa.getPropertyValue("soTimeout"));
 		assertEquals(false, dfa.getPropertyValue("usingDirectBuffers"));
-		
 	}
 
 	@Test
@@ -233,7 +244,6 @@ public class ParserUnitTests {
 		assertEquals(53, dfa.getPropertyValue("soSendBufferSize"));
 		assertEquals(54, dfa.getPropertyValue("soTimeout"));
 		assertEquals(true, dfa.getPropertyValue("usingDirectBuffers"));
-		
 	}
 
 	@Test
@@ -248,7 +258,6 @@ public class ParserUnitTests {
 		assertEquals(27, dfa.getPropertyValue("soTrafficClass"));
 		assertEquals(53, dfa.getPropertyValue("soSendBufferSize"));
 		assertEquals(54, dfa.getPropertyValue("soTimeout"));
-		
 	}
 	
 	@Test
@@ -267,13 +276,13 @@ public class ParserUnitTests {
 		assertEquals(125, dfa.getPropertyValue("soSendBufferSize"));
 		assertEquals(126, dfa.getPropertyValue("soTimeout"));
 		assertEquals(23, dfa.getPropertyValue("poolSize"));
-
+		assertEquals(false, dfa.getPropertyValue("close"));
 	}
 
 	@Test
 	public void testOutGateway() {
 		DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetOutboundGateway);
-		assertTrue(simpleTcpNetOutboundGateway.getPort() >= 6500);
+		assertTrue(simpleTcpNetOutboundGateway.getPort() >= 6600);
 		assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat"));
 		TcpNetSendingMessageHandler handler = (TcpNetSendingMessageHandler) dfa
 				.getPropertyValue("handler");
@@ -284,6 +293,42 @@ public class ParserUnitTests {
 		assertEquals(224, dfa.getPropertyValue("soReceiveBufferSize"));
 		assertEquals(225, delegateDfa.getPropertyValue("soSendBufferSize"));
 		assertEquals(226, delegateDfa.getPropertyValue("soTimeout"));
+		assertEquals(false, dfa.getPropertyValue("close"));
 	}
 
+	@Test
+	public void testInGatewayClose() {
+		DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetInboundGatewayClose);
+		assertTrue(simpleTcpNetInboundGatewayClose.getPort() >= 6700);
+		assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat"));
+		TcpNetReceivingChannelAdapter delegate = (TcpNetReceivingChannelAdapter) dfa
+				.getPropertyValue("delegate");
+		DirectFieldAccessor delegateDfa = new DirectFieldAccessor(delegate);
+		assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReaderClass"));
+		assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass"));
+		assertEquals(true, dfa.getPropertyValue("soKeepAlive"));
+		assertEquals(123, dfa.getPropertyValue("receiveBufferSize"));
+		assertEquals(124, dfa.getPropertyValue("soReceiveBufferSize"));
+		assertEquals(125, dfa.getPropertyValue("soSendBufferSize"));
+		assertEquals(126, dfa.getPropertyValue("soTimeout"));
+		assertEquals(23, dfa.getPropertyValue("poolSize"));
+		assertEquals(true, dfa.getPropertyValue("close"));
+	}
+
+	@Test
+	public void testOutGatewayClose() {
+		DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetOutboundGatewayClose);
+		assertTrue(simpleTcpNetOutboundGatewayClose.getPort() >= 6800);
+		assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat"));
+		TcpNetSendingMessageHandler handler = (TcpNetSendingMessageHandler) dfa
+				.getPropertyValue("handler");
+		DirectFieldAccessor delegateDfa = new DirectFieldAccessor(handler);
+		assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReaderClass"));
+		assertEquals(CustomNetSocketWriter.class, delegateDfa.getPropertyValue("customSocketWriterClass"));
+		assertEquals(true, delegateDfa.getPropertyValue("soKeepAlive"));
+		assertEquals(224, dfa.getPropertyValue("soReceiveBufferSize"));
+		assertEquals(225, delegateDfa.getPropertyValue("soSendBufferSize"));
+		assertEquals(226, delegateDfa.getPropertyValue("soTimeout"));
+		assertEquals(true, dfa.getPropertyValue("close"));
+	}
 }

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

@@ -41,11 +41,14 @@ public class CustomNetSocketReader extends NetSocketReader {
 	 * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat()
 	 */
 	@Override
-	protected boolean assembleDataCustomFormat() throws IOException {
+	protected int assembleDataCustomFormat() throws IOException {
 		byte[] buff = new byte[24];
-		read(buff);
+		int status = read(buff, true);
+		if (status < 0) {
+			return status;
+		}
 		assembledData = buff;
-		return true;
+		return MESSAGE_COMPLETE;
 	}
 	
 	

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

@@ -44,17 +44,17 @@ public class CustomNioSocketReader extends NioSocketReader {
 	 * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat()
 	 */
 	@Override
-	protected boolean assembleDataCustomFormat() throws IOException {
+	protected int assembleDataCustomFormat() throws IOException {
 		if (buffer == null) {
 			buffer = allocate(24);
 		}
 		readChannel(buffer);
 		if (buffer.hasRemaining()) {
-			return false;
+			return MESSAGE_INCOMPLETE;
 		}
 		assembledData = buffer.array();
 		buffer = null;
-		return true;
+		return MESSAGE_COMPLETE;
 	}
 	
 	

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

@@ -45,14 +45,14 @@ public class NetSocketReaderTests {
 		Socket socket = server.accept();
 		socket.setSoTimeout(5000);
 		NetSocketReader reader = new NetSocketReader(socket);
-		if (reader.assembleData()) {
+		if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 			assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 								 new String(reader.getAssembledData()));
 		}
 		else {
 			fail("Failed to assemble first message");
 		}
-		if (reader.assembleData()) {
+		if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 			assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 								 new String(reader.getAssembledData()));
 		}
@@ -75,14 +75,14 @@ public class NetSocketReaderTests {
 		socket.setSoTimeout(5000);
 		NetSocketReader reader = new NetSocketReader(socket);
 		reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
-		if (reader.assembleData()) {
+		if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 			assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 								 new String(reader.getAssembledData()));
 		}
 		else {
 			fail("Failed to assemble first message");
 		}
-		if (reader.assembleData()) {
+		if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 			assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 								 new String(reader.getAssembledData()));
 		}
@@ -105,14 +105,14 @@ public class NetSocketReaderTests {
 		socket.setSoTimeout(5000);
 		NetSocketReader reader = new NetSocketReader(socket);
 		reader.setMessageFormat(MessageFormats.FORMAT_CRLF);
-		if (reader.assembleData()) {
+		if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 			assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 								 new String(reader.getAssembledData()));
 		}
 		else {
 			fail("Failed to assemble first message");
 		}
-		if (reader.assembleData()) {
+		if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 			assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 								 new String(reader.getAssembledData()));
 		}
@@ -131,7 +131,7 @@ public class NetSocketReaderTests {
 		socket.setSoTimeout(5000);
 		NetSocketReader reader = new NetSocketReader(socket);
 		try {
-		    if (reader.assembleData()) {
+		    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 		    	fail("Expected message length exceeded exception");
 		    }
 		} catch (IOException e) {
@@ -153,7 +153,7 @@ public class NetSocketReaderTests {
 		NetSocketReader reader = new NetSocketReader(socket);
 		reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
 		try {
-		    if (reader.assembleData()) {
+		    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 		    	fail("Expected message length exceeded exception");
 		    }
 		} catch (IOException e) {
@@ -176,7 +176,7 @@ public class NetSocketReaderTests {
 		reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
 		reader.setMaxMessageSize(1024);
 		try {
-		    if (reader.assembleData()) {
+		    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 		    	fail("Expected message length exceeded exception");
 		    }
 		} catch (IOException e) {
@@ -198,7 +198,7 @@ public class NetSocketReaderTests {
 		NetSocketReader reader = new NetSocketReader(socket);
 		reader.setMessageFormat(MessageFormats.FORMAT_CRLF);
 		try {
-		    if (reader.assembleData()) {
+		    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 		    	fail("Expected message length exceeded exception");
 		    }
 		} catch (IOException e) {
@@ -221,7 +221,7 @@ public class NetSocketReaderTests {
 		reader.setMessageFormat(MessageFormats.FORMAT_CRLF);
 		reader.setMaxMessageSize(1024);
 		try {
-		    if (reader.assembleData()) {
+		    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 		    	fail("Expected message length exceeded exception");
 		    }
 		} catch (IOException e) {

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

@@ -82,7 +82,7 @@ public class NioSocketReaderTests {
 				iterator.remove();
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
-					if (reader.assembleData()) {
+					if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 						assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 						         new String(reader.getAssembledData()));
 						count++;
@@ -138,7 +138,7 @@ public class NioSocketReaderTests {
 				iterator.remove();
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
-					if (reader.assembleData()) {
+					if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 						assertEquals("Data", "xx", 
 						         new String(reader.getAssembledData()));
 						done = true;
@@ -198,7 +198,7 @@ public class NioSocketReaderTests {
 				iterator.remove();
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
-					if (reader.assembleData()) {
+					if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 						assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 						         new String(reader.getAssembledData()));
 						count++;
@@ -258,7 +258,7 @@ public class NioSocketReaderTests {
 				iterator.remove();
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
-					if (reader.assembleData()) {
+					if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 						assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 						         new String(reader.getAssembledData()));
 						count++;
@@ -318,7 +318,7 @@ public class NioSocketReaderTests {
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
 					try {
-					    if (reader.assembleData()) {
+					    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 					    	fail("Expected message length exceeded exception");
 					    }
 					} catch (IOException e) {
@@ -387,7 +387,7 @@ public class NioSocketReaderTests {
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
 					try {
-					    if (reader.assembleData()) {
+					    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 					    	fail("Expected message length exceeded exception");
 					    }
 					} catch (IOException e) {
@@ -456,7 +456,7 @@ public class NioSocketReaderTests {
 				if (key.isReadable()) {
 					assertEquals(channel, key.channel());
 					try {
-					    if (reader.assembleData()) {
+					    if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) {
 					    	fail("Expected message length exceeded exception");
 					    }
 					} catch (IOException e) {

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

@@ -134,9 +134,9 @@ public class SimpleTcpNetInboundGatewayTests {
 		startup = 0;
 		Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayCustom.getPort());
 		String greetings = "Hello World!";
-		String pad = "                        ";
+		String pad = "            ";
 		socket.getOutputStream().write((greetings).getBytes());
-		socket.getOutputStream().write(pad.getBytes()); // will be truncated
+		socket.getOutputStream().write(pad.getBytes());
 		StringBuilder sb = new StringBuilder();
 		int c;
 		int n = 0;

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.integration.ip.tcp;
 
+import static org.junit.Assert.assertEquals;
+
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.concurrent.CountDownLatch;
@@ -22,7 +24,6 @@ import java.util.concurrent.TimeUnit;
 
 import javax.net.ServerSocketFactory;
 
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -35,8 +36,6 @@ import org.springframework.integration.message.MessageBuilder;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import static org.junit.Assert.assertEquals;
-
 /**
  * @author Gary Russell
  *
@@ -74,16 +73,29 @@ public class SimpleTcpNetOutboundGatewayTests {
 		SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
 			("localhost", inboundGatewayCrLf.getPort());
 		gateway.setMessageFormat(MessageFormats.FORMAT_CRLF);
+		waitListening(inboundGatewayCrLf);
 		Message<String> message = MessageBuilder.withPayload("test").build();
 		byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
 		assertEquals("echo:test", new String(bytes));
 	}
 
+	private void waitListening(SimpleTcpNetInboundGateway gateway) throws Exception {
+		int n = 0;
+		while (!gateway.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 100) {
+				throw new Exception("Gateway failed to listen");
+			}
+		}
+		
+	}
+
 	@Test
 	public void testOutboundStxEtx() throws Exception {
 		SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
 			("localhost", inboundGatewayStxEtx.getPort());
 		gateway.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
+		waitListening(inboundGatewayStxEtx);
 		Message<String> message = MessageBuilder.withPayload("test").build();
 		byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
 		assertEquals("echo:test", new String(bytes));
@@ -94,6 +106,7 @@ public class SimpleTcpNetOutboundGatewayTests {
 		SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
 			("localhost", inboundGatewayLength.getPort());
 		gateway.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER);
+		waitListening(inboundGatewayLength);
 		Message<String> message = MessageBuilder.withPayload("test").build();
 		byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
 		assertEquals("echo:test", new String(bytes));
@@ -106,6 +119,7 @@ public class SimpleTcpNetOutboundGatewayTests {
 		gateway.setMessageFormat(MessageFormats.FORMAT_CUSTOM);
 		gateway.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader");
 		gateway.setCustomSocketWriterClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter");
+		waitListening(inboundGatewayCustom);
 		Message<String> message = MessageBuilder.withPayload("test").build();
 		byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
 		assertEquals("echo:test", new String(bytes).trim());
@@ -119,11 +133,12 @@ public class SimpleTcpNetOutboundGatewayTests {
 		assertEquals("echo:test", new String(bytes).trim());
 	}
 	
-	@Ignore @Test
+	@Test
 	public void testOutboundClose() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
+		final CountDownLatch latch3 = new CountDownLatch(1);
 		Thread t = new Thread(new Runnable() {
 			public void run() {
 				try {
@@ -134,8 +149,9 @@ public class SimpleTcpNetOutboundGatewayTests {
 						byte[] b = new byte[1024];
 						s.getInputStream().read(b);
 						s.getOutputStream().write("OK\r\n".getBytes());
-						s.close();
 						latch2.countDown();
+						latch3.await();
+						s.close();
 					}
 				} catch (Exception e) {
 					e.printStackTrace();
@@ -146,9 +162,11 @@ public class SimpleTcpNetOutboundGatewayTests {
 		SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
 			("localhost", port);
 		gateway.setMessageFormat(MessageFormats.FORMAT_CRLF);
+		gateway.setClose(true);
 		Message<String> message = MessageBuilder.withPayload("test").build();
 		byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
 		assertEquals("OK", new String(bytes));
+		latch3.countDown();
 		latch2.await(2000, TimeUnit.MILLISECONDS);
 		bytes = (byte[]) gateway.handleRequestMessage(message);
 		assertEquals("OK", new String(bytes));

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

@@ -91,8 +91,8 @@ public class SocketMessageMapperTests {
 		/* (non-Javadoc)
 		 * @see org.springframework.integration.ip.tcp.SocketReader#assembleData()
 		 */
-		public boolean assembleData() {
+		public int assembleData() {
-			return false;
+			return SocketReader.MESSAGE_INCOMPLETE;
 		}
 
 		/* (non-Javadoc)

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

@@ -18,6 +18,8 @@ package org.springframework.integration.ip.tcp;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
+import java.util.concurrent.CountDownLatch;
+
 import org.junit.Test;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.Message;
@@ -44,14 +46,13 @@ public class TcpReceivingChannelAdapterTests {
 		taskScheduler.initialize();
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.start();
-		Thread.sleep(2000); // wait for server to start listening
+		waitListening(adapter);
 		SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice
-		Thread.sleep(2000); // wait for asynch processing
+		Message<?> message = channel.receive(2000);
-		Message<?> message = channel.receive(0);
 		assertNotNull(message);
 		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 				new String((byte[])message.getPayload()));
-		message = channel.receive(0);
+		message = channel.receive(2000);
 		assertNotNull(message);
 		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 				new String((byte[])message.getPayload()));
@@ -74,14 +75,13 @@ public class TcpReceivingChannelAdapterTests {
 		taskScheduler.initialize();
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.start();
-		Thread.sleep(2000); // wait for server to start listening
+		waitListening(adapter);
 		SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice
-		Thread.sleep(4000); // wait for asynch processing
+		Message<?> message = channel.receive(4000);
-		Message<?> message = channel.receive(0);
 		assertNotNull(message);
 		assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", 
 				new String((byte[])message.getPayload()));
-		message = channel.receive(0);
+		message = channel.receive(2000);
 		assertNotNull(message);
 		assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", 
 				new String((byte[])message.getPayload()));
@@ -102,14 +102,13 @@ public class TcpReceivingChannelAdapterTests {
 		taskScheduler.initialize();
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.start();
-		Thread.sleep(2000); // wait for server to start listening
+		waitListening(adapter);
 		SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice
-		Thread.sleep(2000); // wait for asynch processing
+		Message<?> message = channel.receive(2000);
-		Message<?> message = channel.receive(0);
 		assertNotNull(message);
 		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 				new String((byte[])message.getPayload()));
-		message = channel.receive(0);
+		message = channel.receive(2000);
 		assertNotNull(message);
 		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
 				new String((byte[])message.getPayload()));
@@ -131,18 +130,98 @@ public class TcpReceivingChannelAdapterTests {
 		taskScheduler.initialize();
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.start();
-		Thread.sleep(2000); // wait for server to start listening
+		waitListening(adapter);
 		SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice
-		Thread.sleep(4000); // wait for asynch processing
+		Message<?> message = channel.receive(2000);
-		Message<?> message = channel.receive(0);
 		assertNotNull(message);
 		assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", 
 				new String((byte[])message.getPayload()));
-		message = channel.receive(0);
+		message = channel.receive(2000);
 		assertNotNull(message);
 		assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", 
 				new String((byte[])message.getPayload()));
 		adapter.stop();
 	}
 
+	/**
+	 * Tests close option on inbound adapter.
+	 * 
+	 * @throws Exception
+	 */
+	@Test
+	public void testNetClose() throws Exception {
+		QueueChannel channel = new QueueChannel(2);
+		int port = SocketUtils.findAvailableServerSocket();
+		AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port);
+		adapter.setOutputChannel(channel);
+		adapter.setClose(true);
+		adapter.setMessageFormat(MessageFormats.FORMAT_CRLF);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.start();
+		waitListening(adapter);
+		CountDownLatch latch = new CountDownLatch(1);
+		SocketUtils.testSendCrLfSingle(port, latch); 
+		Message<?> message = channel.receive(5000);
+		latch.countDown();
+		assertNotNull(message);
+		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+		latch = new CountDownLatch(1);
+		SocketUtils.testSendCrLfSingle(port, latch); 
+		message = channel.receive(5000);
+		latch.countDown();
+		assertNotNull(message);
+		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+		adapter.stop();
+	}
+
+	/**
+	 * Tests close option on inbound adapter.
+	 * 
+	 * @throws Exception
+	 */
+	@Test
+	public void testNioClose() throws Exception {
+		QueueChannel channel = new QueueChannel(2);
+		int port = SocketUtils.findAvailableServerSocket();
+		AbstractTcpReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port);
+		adapter.setOutputChannel(channel);
+		adapter.setClose(true);
+		adapter.setMessageFormat(MessageFormats.FORMAT_CRLF);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.start();
+		waitListening(adapter);
+		CountDownLatch latch = new CountDownLatch(1);
+		SocketUtils.testSendCrLfSingle(port, latch); 
+		Message<?> message = channel.receive(2000);
+		assertNotNull(message);
+		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+		latch = new CountDownLatch(1);
+		SocketUtils.testSendCrLfSingle(port, latch); 
+		message = channel.receive(2000);
+		assertNotNull(message);
+		assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+		adapter.stop();
+	}
+	
+	private void waitListening(AbstractInternetProtocolReceivingChannelAdapter adapter) throws Exception {
+		int n = 0;
+		while (!adapter.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 100) {
+				throw new Exception("Gateway failed to listen");
+			}
+		}
+		
+	}
+
+	
+
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java

@@ -222,7 +222,34 @@ public class SocketUtils {
 		thread.setDaemon(true);
 		thread.start();
 	}
-	
+
+	/**
+	 * Sends a single message +CRLF.
+	 * @param latch Waits for latch to count down before closing the socket.
+	 */
+	public static void testSendCrLfSingle(final int port, final CountDownLatch latch) {
+		Thread thread = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = new Socket(InetAddress.getByName("localhost"), port);
+					OutputStream outputStream = socket.getOutputStream();
+					outputStream.write(TEST_STRING.getBytes());
+					outputStream.write(TEST_STRING.getBytes());
+					writeByte(outputStream, '\r', true);
+					writeByte(outputStream, '\n', true);
+					if (latch != null) {
+						latch.await();
+					}
+					socket.close();
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		thread.setDaemon(true);
+		thread.start();
+	}
+
 	/**
 	 * Sends a large CRLF message with no CRLF.
 	 */