Inbound tcp channel adapters; no outbound yet, nor namespace support.

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

@@ -34,7 +34,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
 
 	protected final int port;
 
-	protected volatile int soTimeout = 60 * 1000;
+	protected volatile int soTimeout = 0;
 
 	protected volatile int soReceiveBufferSize = -1;
 

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

@@ -0,0 +1,102 @@
+/*
+ * 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;
+
+import java.io.IOException;
+
+/**
+ * Abstract SocketReader that handles data in 3 standard, and one custom
+ * format. The default format is {@link MessageFormats#FORMAT_LENGTH_HEADER} in which 
+ * the message consists of a 4 byte integer (in network byte order) containing
+ * the length of data that follows. {@link MessageFormats#FORMAT_STX_ETX} 
+ * indicates a message where the data begins with STX (0x02) and ends with 
+ * ETX (0x03); the STX and ETX are not part of the data. {@link MessageFormats#FORMAT_CRLF}
+ * indicates a message followed by carriage return and line feed '\r\n'. 
+ * FORMAT_LENGTH_HEADER can be used for {@link java.net.Socket} and 
+ * {@link java.nio.channels.SocketChannel} implementations are provided for
+ * the standard formats. Users requiring other formats should subclass the
+ * appropriate implementation, and provide an implementation for 
+ * {@link #assembleDataCustomFormat()} which is invoked by {@link #assembleData()}
+ * when the format is {@link MessageFormats#FORMAT_CUSTOM}. 
+ * 
+ * @author Gary Russell
+ *
+ */
+public abstract class AbstractSocketReader implements SocketReader, MessageFormats {
+
+	protected int messageFormat = FORMAT_LENGTH_HEADER;
+
+	/**
+	 * The assembled data; must contain a reference when assembleData()
+	 * returns true; will be set to null when getAssembledData() is called.
+	 */
+	protected byte[] assembledData;
+	
+	/**
+	 * Assembles data in format {@link #FORMAT_LENGTH_HEADER}.
+	 * @return True when a message is completely assembled.
+	 * @throws IOException
+	 */
+	protected abstract boolean 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;
+	
+	/**
+	 * Assembles data in format {@link #FORMAT_CRLF}.
+	 * @return True when a message is completely assembled.
+	 * @throws IOException
+	 */
+	protected abstract boolean assembleDataCrLfFormat() throws IOException;
+	
+	/**
+	 * Assembles data in format {@link #FORMAT_CUSTOM}. Implementations must
+	 * return false until the message is completely assembled, at which time
+	 * the implementation must update assembledData to reference the assembled
+	 * message.
+	 * @return True when a message is completely assembled.
+	 * @throws IOException
+	 */
+	protected abstract boolean assembleDataCustomFormat() throws IOException;
+	
+	public boolean assembleData() throws IOException {
+		switch (this.messageFormat) {
+		case FORMAT_LENGTH_HEADER:
+			return assembleDataLengthFormat();
+		case FORMAT_STX_ETX:
+			return assembleDataStxEtxFormat();
+		case FORMAT_CRLF:
+			return assembleDataCrLfFormat();
+		case FORMAT_CUSTOM:
+			return assembleDataCustomFormat();
+		default:
+			throw new UnsupportedOperationException(
+					"Unsupported message format: " + messageFormat);
+		}
+	}
+
+	/**
+	 * @param messageFormat the messageFormat to set,
+	 */
+	public void setMessageFormat(int messageFormat) {
+		this.messageFormat = messageFormat;
+	}
+
+}

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

@@ -0,0 +1,150 @@
+/*
+ * 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;
+
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.concurrent.ThreadFactory;
+
+import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * Abstract class for tcp/ip incoming channel adapters. Implementations
+ * for {@link java.net.Socket} and {@link java.nio.channels.SocketChannel}
+ * are provided.
+ * 
+ * @author Gary Russell
+ *
+ */
+public abstract class AbstractTcpReceivingChannelAdapter extends
+		AbstractInternetProtocolReceivingChannelAdapter {
+
+	protected volatile ThreadPoolTaskScheduler threadPoolTaskScheduler;
+
+	protected volatile int poolSize = -1;
+	
+	protected volatile SocketMessageMapper mapper = new SocketMessageMapper();
+
+	protected volatile boolean soKeepAlive;
+
+	protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
+	
+	protected Class<SocketReader> customSocketReader;
+	
+	protected boolean usingDirectBuffers; 
+	
+	/**
+	 * Constructs a receiving channel adapter that listens on the port.
+	 * @param port The port to listen on.
+	 */
+	public AbstractTcpReceivingChannelAdapter(int port) {
+		super(port);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.endpoint.AbstractEndpoint#doStop()
+	 */
+	@Override
+	protected void doStop() {
+		// TODO Auto-generated method stub
+
+	}
+
+	/**
+	 * Creates the ThreadPoolTaskScheduler, if necessary, and calls 
+	 * {@link #server()}.
+	 */
+	public void run() {
+		if (logger.isDebugEnabled()) {
+			logger.debug(this.getClass().getSimpleName() + " running...");
+		}
+		if (this.active && this.threadPoolTaskScheduler == null) {
+			this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
+			this.threadPoolTaskScheduler.setThreadFactory(new ThreadFactory() {
+				public Thread newThread(Runnable runner) {
+					Thread thread = new Thread(runner);
+					thread.setName("TCP-Incoming-Msg-Handler");
+					thread.setDaemon(true);
+					return thread;
+				}
+			});
+			if (this.poolSize > 0) {
+				this.threadPoolTaskScheduler.setPoolSize(this.poolSize);
+			}
+			this.threadPoolTaskScheduler.initialize();
+		}
+		server();
+	}
+
+	/**
+	 * Establishes the server.
+	 */
+	protected abstract void server();
+
+	/**
+	 * Sets soTimeout, soKeepAlive and tcpNoDelay according to the configured
+	 * properties.
+	 * @param socket The socket.
+	 * @throws SocketException
+	 */
+	protected void setSocketOptions(Socket socket) throws SocketException {
+		socket.setSoTimeout(this.soTimeout);
+		if (this.soReceiveBufferSize > 0) {
+			socket.setReceiveBufferSize(this.soReceiveBufferSize);
+		}
+		socket.setKeepAlive(this.soKeepAlive);
+		
+	}
+
+	/**
+	 * @param soKeepAlive the soKeepAlive to set
+	 */
+	public void setSoKeepAlive(boolean soKeepAlive) {
+		this.soKeepAlive = soKeepAlive;
+	}
+
+	/**
+	 * @param messageFormat the messageFormat to set
+	 */
+	public void setMessageFormat(int messageFormat) {
+		this.messageFormat = messageFormat;
+	}
+
+	/**
+	 * @param customSocketReader the customSocketReader to set
+	 * @throws ClassNotFoundException 
+	 */
+	@SuppressWarnings("unchecked")
+	public void setCustomSocketReader(String customSocketReaderClassName) throws ClassNotFoundException {
+		this.customSocketReader = (Class<SocketReader>) Class.forName(customSocketReaderClassName);
+	}
+
+	/**
+	 * @param usingDirectBuffers the usingDirectBuffers to set
+	 */
+	public void setUsingDirectBuffers(boolean usingDirectBuffers) {
+		this.usingDirectBuffers = usingDirectBuffers;
+	}
+
+	/**
+	 * @param poolSize the poolSize to set
+	 */
+	public void setPoolSize(int poolSize) {
+		this.poolSize = poolSize;
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java

@@ -0,0 +1,47 @@
+/*
+ * 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;
+
+/**
+ * Definition of message formats supported by tcp channel adapters.
+ * 
+ * @author Gary Russell
+ *
+ */
+public interface MessageFormats {
+
+	/**
+	 * Message has format '<length><message>'.
+	 */
+	public static final int FORMAT_LENGTH_HEADER = 1;
+	/**
+	 * Message has format 'STX<message>ETX'.
+	 */
+	public static final int FORMAT_STX_ETX = 2;
+	/**
+	 * Message has format '<message>\r\n'.
+	 */
+	public static final int FORMAT_CRLF = 3;
+	/**
+	 * Message has custom format.
+	 */
+	public static final int FORMAT_CUSTOM = 99;
+	
+	public static final int STX = 0x02;
+	
+	public static final int ETX = 0x03;
+
+}

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

@@ -0,0 +1,170 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.adapter.MessageMappingException;
+
+/**
+ * A SocketReader that reads from a {@link java.net.Socket}. Threads
+ * calling {@link NetSocketReader#assembledData} will block until a message
+ * is completely assembled.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class NetSocketReader extends AbstractSocketReader {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+	
+	protected Socket socket;
+
+	protected int receiveBufferSize = 1024 * 60;
+	
+	/**
+	 * Constructs a NetsocketReader which reads from the Socket.
+	 * @param socket The socket.
+	 */
+	public NetSocketReader(Socket socket) {
+		this.socket = socket;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#read(java.nio.ByteBuffer)
+	 */
+	@Override
+	protected boolean assembleDataLengthFormat() throws IOException {
+		byte[] lengthPart = new byte[4];
+		read(lengthPart);
+		int messageLength = ByteBuffer.wrap(lengthPart).getInt();
+		if (logger.isDebugEnabled()) {
+			logger.debug("Message length is " + messageLength);
+		}		
+		byte[] messagePart = new byte[messageLength];
+		read(messagePart);
+		assembledData = messagePart;
+		return true;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat()
+	 */
+	@Override
+	protected boolean assembleDataStxEtxFormat() throws IOException {
+		InputStream inputStream = socket.getInputStream();
+		if (inputStream.read() != STX)
+			throw new MessageMappingException("Expected STX to begin message");
+		byte[] buffer = new byte[receiveBufferSize];
+		int n = 0;
+		int bite;
+		while ((bite = inputStream.read()) != ETX) {
+			buffer[n++] = (byte) bite;
+		}
+		assembledData = new byte[n];
+		System.arraycopy(buffer, 0, assembledData, 0, n);
+		return true;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat()
+	 */
+	@Override
+	protected boolean assembleDataCrLfFormat() throws IOException {
+		InputStream inputStream = socket.getInputStream();
+		byte[] buffer = new byte[receiveBufferSize];
+		int n = 0;
+		int bite;
+		while (true) {
+			bite = inputStream.read();
+			if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
+				break;
+			buffer[n++] = (byte) bite;
+		};
+		assembledData = new byte[n-1];
+		System.arraycopy(buffer, 0, assembledData, 0, n-1);
+		return true;
+	}
+
+	/**
+	 * Throws {@link UnsupportedOperationException}; custom implementations can
+	 * subclass this class and provide an implementation.
+	 * @throws IOException 
+	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat().
+	 * 
+	 */
+	@Override
+	protected boolean assembleDataCustomFormat() throws IOException {
+		throw new UnsupportedOperationException("Need to subclass for this format");
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#getAssembledData()
+	 */
+	public byte[] getAssembledData() {
+		byte[] assembledData = this.assembledData;
+		this.assembledData = null;
+		return assembledData;
+	}
+
+	/**
+	 * Reads data from the socket and puts the data in buffer. Blocks until
+	 * buffer is full or a socket timeout occurs.
+	 * @param buffer
+	 * @throws IOException
+	 */
+	protected void read(byte[] buffer) 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) {
+				throw new IOException("EOF");
+			}
+			lengthRead += len;
+			if (logger.isDebugEnabled()) {
+				logger.debug("Read " + len + " bytes, buffer is now at " + 
+							 lengthRead + " of " +
+							 needed);
+			}
+		}
+		
+	}
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#getAddress()
+	 */
+	public InetAddress getAddress() {
+		return this.socket.getInetAddress();
+	}
+
+	/**
+	 * Sets the socket.
+	 * @param socket
+	 */
+	public void setSocket(Socket socket) {
+		this.socket = socket;
+	}
+
+}

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

@@ -0,0 +1,319 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.adapter.MessageMappingException;
+
+/**
+ * A non-blocking SocketReader that reads from a {@link java.nio.channels.SocketChannel}.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class NioSocketReader extends AbstractSocketReader {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+	
+	protected SocketChannel channel;
+	
+	protected boolean usingDirectBuffers;
+	
+	protected ByteBuffer lengthPart;
+	
+	protected ByteBuffer dataPart;
+	
+	protected ByteBuffer rawBuffer;
+	
+	protected ByteBuffer buildBuffer;
+
+	protected int receiveBufferSize = 1024 * 60;
+
+	protected boolean building;
+
+	/**
+	 * Constructs an NioSocketReader which reads from the SocketChannel.
+	 * @param channel The channel.
+	 */
+	public NioSocketReader(SocketChannel channel) {
+		this.channel = channel;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#read(java.nio.ByteBuffer, int)
+	 */
+	public byte[] getAssembledData() {
+		byte[] assembledData = this.assembledData;
+		this.assembledData = null;
+		return assembledData;
+	}
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#assembleData()
+	 */
+	@Override
+	public boolean assembleDataLengthFormat() {
+		try {
+			if (lengthPart == null) {
+				lengthPart = allocate(4);
+			}
+			if (lengthPart.hasRemaining()) {
+				readChannel(lengthPart);
+				return false;
+			}
+			if (dataPart == null) {
+				lengthPart.flip();
+				int messageLength = lengthPart.getInt();
+				if (logger.isDebugEnabled()) {
+					logger.debug("Message length is " + messageLength);
+				}
+				dataPart = allocate(messageLength);
+			}
+			if (dataPart.hasRemaining()) {
+				readChannel(dataPart);
+				if (dataPart.hasRemaining()) {
+					return false;
+				}
+			} 
+			assembledData = dataPart.array();
+			lengthPart = dataPart = null;
+			return true;
+		} catch (Exception e) {
+			e.printStackTrace();
+			// TODO
+			try {
+				channel.close();
+			} catch (IOException e1) { 
+				// TODO Auto-generated catch block
+				e1.printStackTrace();
+			}
+			throw new MessageMappingException("Message assembly exception", e);
+		}
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat()
+	 */
+	@Override
+	protected boolean assembleDataStxEtxFormat() throws IOException {
+		if (readChannelNonDeterministic()) {
+			byte bite = rawBuffer.get();
+			int count = 0;
+			if (!building) { 
+				if (bite != STX) {
+					throw new MessageMappingException("Expected STX, received " + Integer.toHexString(bite));
+				}
+				building = true;
+				count++;
+				if (!rawBuffer.hasRemaining()) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Incomplete message, consumed 1 byte");
+					}
+					return false;
+				}
+			} else {
+				if (bite == ETX) {
+					finishAssembly();
+					return true;
+				}
+				buildBuffer.put(bite);
+				count++;
+			}
+			while (true) {
+				if (!rawBuffer.hasRemaining()) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Incomplete message, consumed " + count + " bytes");
+					}
+					return false;
+				}
+				bite = rawBuffer.get();
+				if (bite == ETX) {
+					break;
+				}
+				buildBuffer.put(bite);
+				count++;
+			} 
+			if (logger.isDebugEnabled()) {
+				logger.debug("Consumed " + count + " bytes");
+			}
+			finishAssembly();
+			return true;
+		} else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Incomplete message, consumed 0 bytes");
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * 
+	 */
+	private void finishAssembly() {
+		assembledData = new byte[buildBuffer.position()];
+		System.arraycopy(buildBuffer.array(), 0, assembledData, 0, assembledData.length);
+		building = false;
+		buildBuffer.clear();
+		logger.debug("Message assembly complete");
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat()
+	 */
+	@Override
+	protected boolean assembleDataCrLfFormat() throws IOException {
+		if (readChannelNonDeterministic()) {
+			int count = 0;
+			while (true) {
+				if (!rawBuffer.hasRemaining()) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Incomplete message, consumed " + count + " bytes");
+					}
+					return false;
+				}
+				byte bite = rawBuffer.get();
+				if (bite == '\n' && buildBuffer.position() > 0) {
+					buildBuffer.position(buildBuffer.position() - 1);
+					if (buildBuffer.get() == '\r') {
+						buildBuffer.position(buildBuffer.position() - 1);
+						break;
+					}
+				}
+				buildBuffer.put(bite);
+				count++;
+			} 
+			if (logger.isDebugEnabled()) {
+				logger.debug("Consumed " + count + " bytes");
+			}
+			finishAssembly();
+			return true;
+		} else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Incomplete message, consumed 0 bytes");
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * Throws {@link UnsupportedOperationException}; custom implementations can
+	 * subclass this class and provide an implementation.
+	 * @throws IOException 
+	 * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat().
+	 * 
+	 */
+	@Override
+	protected boolean assembleDataCustomFormat() throws IOException {
+		throw new UnsupportedOperationException("Need to subclass for this format");
+	}
+
+	/**
+	 * Reads from the channel into the buffer. Reads as much data as is 
+	 * currently available in the channel.
+	 * @param buffer
+	 * @throws IOException
+	 */
+	protected void readChannel(ByteBuffer buffer) throws IOException {
+		try {
+			int len = channel.read(buffer);
+			if (logger.isDebugEnabled()) {
+				logger.debug("Read " + len + " bytes, buffer is now at " + 
+							 buffer.position() + " of " +
+							 buffer.capacity());
+			}
+		} catch (IOException e) {
+			throw e;
+		}
+	}
+	
+	/**
+	 * Reads data into the rawBuffer for non-deterministic algorithms.
+	 * @return true If data is available.
+	 * @throws IOException
+	 */
+	protected boolean readChannelNonDeterministic() throws IOException {
+		if (rawBuffer == null) {
+			rawBuffer = allocate(receiveBufferSize);
+			buildBuffer = ByteBuffer.allocate(receiveBufferSize);
+		} else if (rawBuffer.hasRemaining()) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Raw buffer has " + rawBuffer.remaining() + " remaining");
+			}
+			return true;
+		}
+		rawBuffer.clear();
+		int len = channel.read(rawBuffer);
+		if (len == 0) {
+			return false;
+		}
+		rawBuffer.flip();
+		if (logger.isDebugEnabled()) {
+			logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
+		}
+		return true;
+	}
+	
+	/**
+	 * Allocates a ByteBuffer of the requested length using normal or
+	 * direct buffers, depending on the usingDirectBuffers field.
+	 * @param length 
+	 * @return
+	 */
+	protected ByteBuffer allocate(int length) {
+		ByteBuffer buffer;
+		if (usingDirectBuffers) {
+			buffer = ByteBuffer.allocateDirect(length);
+		} else {
+			buffer = ByteBuffer.allocate(length);
+		}
+		return buffer;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#getAddress()
+	 */
+	public InetAddress getAddress() {
+		return this.channel.socket().getInetAddress();
+	}
+	
+	/**
+	 * @return the useDirectBuffers
+	 */
+	public boolean isUsingDirectBuffers() {
+		return usingDirectBuffers;
+	}
+
+	/**
+	 * @param useDirectBuffers the useDirectBuffers to set
+	 */
+	public void setUsingDirectBuffers(boolean usingDirectBuffers) {
+		this.usingDirectBuffers = usingDirectBuffers;
+	}
+
+	/**
+	 * @param channel
+	 */
+	public void setChannel(SocketChannel channel) {
+		this.channel = channel;
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java

@@ -0,0 +1,70 @@
+/*
+ * 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;
+
+import java.io.IOException;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.message.InboundMessageMapper;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.OutboundMessageMapper;
+
+/**
+ * Maps incoming data from a {@link SocketReader} to a {@link Message} and from
+ * a Message to outgoing data forwarded to a {@link SocketWriter}.
+ * @author Gary Russell
+ *
+ */
+public class SocketMessageMapper implements
+		InboundMessageMapper<SocketReader>, 
+		OutboundMessageMapper<SocketWriter> {
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.message.InboundMessageMapper#toMessage(java.lang.Object)
+	 */
+	public Message<byte[]> toMessage(SocketReader socketReader) throws Exception {
+		return fromRaw(socketReader);
+	}
+
+
+	/**
+	 * @param socketReader
+	 * @return
+	 * @throws IOException 
+	 */
+	private Message<byte[]> fromRaw(SocketReader socketReader) throws IOException {
+		byte[] payload = socketReader.getAssembledData();
+		Message<byte[]> message = null;
+		if (payload != null && payload.length > 0) {
+			message = MessageBuilder.withPayload(payload)
+					.setHeader(IpHeaders.HOSTNAME, socketReader.getAddress().getHostName())
+					.setHeader(IpHeaders.IP_ADDRESS, socketReader.getAddress().getHostAddress())
+					.build();
+		}
+		return message;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.message.OutboundMessageMapper#fromMessage(org.springframework.integration.core.Message)
+	 */
+	public SocketWriter fromMessage(Message<?> message) throws Exception {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.net.InetAddress;
+
+/**
+ * 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
+ *
+ */
+public interface SocketReader {
+
+	/**
+	 * 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.
+	 * @throws IOException 
+	 */
+	public boolean assembleData() throws IOException;
+	
+	/**
+	 * Retrieves the assembled tcp data or null if the data is not
+	 * yet assembled. Once this method is called, the assembled data is
+	 * again null until a new assembly is completed.
+	 * @return The assembled data or null.
+	 */
+	public byte[] getAssembledData();
+
+	/**
+	 * Returns the InetAddress of the underlying socket.
+	 * @return The InetAddress.
+	 */
+	public InetAddress getAddress();
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.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;
+
+
+/**
+ * A general interface for writing to sockets.
+ * 
+ * @author Gary Russell
+ *
+ */
+public interface SocketWriter {
+
+}

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

@@ -0,0 +1,120 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+
+import javax.net.ServerSocketFactory;
+
+import org.springframework.integration.adapter.MessageMappingException;
+import org.springframework.integration.core.Message;
+
+/**
+ * Tcp Receiving Channel adapter that uses a {@link java.net.Socket}. Each
+ * connected socket uses a dedicated thread so the pool size must be set
+ * accordingly.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class TcpNetReceivingChannelAdapter extends
+		AbstractTcpReceivingChannelAdapter {
+
+	/**
+	 * Constructs a TcpNetReceivingChannelAdapter that listens on the port.
+	 * @param port The port.
+	 */
+	public TcpNetReceivingChannelAdapter(int port) {
+		super(port);
+	}
+
+	/**
+	 * Creates the server socket, listens for incoming connections and schedules
+	 * execution of the {@link #handleSocket(Socket)} method for each new
+	 * connection.
+	 * 
+	 * @see org.springframework.integration.ip.tcp.AbstractTcpReceivingChannelAdapter#server()
+	 */
+	@Override
+	protected void server() {
+		while (true) {
+			try {
+				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+				while (true) {
+					final Socket socket = server.accept();
+					setSocketOptions(socket);
+					this.threadPoolTaskScheduler.execute(new Runnable() {
+						public void run() {
+							handleSocket(socket);
+						}});
+				}
+			} catch (IOException e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+
+		}
+
+	}
+
+	/**
+	 * Constructs a {@link NetSocketReader} and calls its {@link NetSocketReader#assembledData}
+	 * method repeatedly; for each assembled message, calls {@link #sendMessage(Message)} with
+	 * the mapped message.
+	 * 
+	 * @param socket
+	 */
+	protected void handleSocket(Socket socket) {
+		NetSocketReader reader = null;
+		if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
+			try {
+				reader = (NetSocketReader) customSocketReader.newInstance();
+				reader.setSocket(socket);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				throw new MessageMappingException("Failed to instantiate custom reader", e);
+			}
+		}
+		else {
+			reader = new NetSocketReader(socket);
+		}
+		reader.setMessageFormat(messageFormat);
+		while (true) {
+			try {
+				if (reader.assembleData()) {
+					Message<byte[]> message = mapper.toMessage(reader);
+					if (message != null) {
+						sendMessage(message);
+					}
+				}
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				try {
+					socket.close();
+					return;
+				} catch (IOException e1) {
+					// TODO Auto-generated catch block
+					e1.printStackTrace();
+				}
+			}
+		}
+	}
+
+}

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

@@ -0,0 +1,188 @@
+/*
+ * 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;
+
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketException;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.Iterator;
+import java.util.Set;
+
+import org.springframework.integration.core.Message;
+
+/**
+ * Tcp Receiving Channel adapter that uses a {@link java.nio.channels.SocketChannel}.
+ * Sockets are multiplexed across the pooled threads. More than one thread will
+ * be required with large numbers of connections and incoming traffic.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class TcpNioReceivingChannelAdapter extends
+		AbstractTcpReceivingChannelAdapter {
+
+	/**
+	 * Constructs a TcpNioReceivingChannelAdapter to listen on the port.
+	 * @param port The port.
+	 */
+	public TcpNioReceivingChannelAdapter(int port) {
+		super(port);
+	}
+
+	/**
+	 * Opens a non-blocking {@link ServerSocketChannel}, registers it with a 
+	 * {@link Selector} and calls {@link #doSelect(ServerSocketChannel, Selector)}.
+	 * 
+	 * @see org.springframework.integration.ip.tcp.AbstractTcpReceivingChannelAdapter#server()
+	 */
+	@Override
+	protected void server() {
+		try {
+			final ServerSocketChannel server = ServerSocketChannel.open();
+			server.configureBlocking(false);
+			server.socket().bind(new InetSocketAddress(port));
+			final Selector selector = Selector.open();
+			server.register(selector, SelectionKey.OP_ACCEPT);
+			doSelect(server, selector);
+
+		} catch (IOException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	/**
+	 * Listens for incoming connections and for notifications that a connected
+	 * socket is ready for reading.
+	 * Accepts incoming connections, registers the new socket with the 
+	 * selector for reading.
+	 * When a socket is ready for reading, unregisters the read interest and
+	 * schedules a call to doRead which reads all available data. When the read
+	 * is complete, the socket is again registered for read interest. 
+	 * @param server
+	 * @param selector
+	 * @throws IOException
+	 * @throws ClosedChannelException
+	 * @throws SocketException
+	 */
+	private void doSelect(ServerSocketChannel server, final Selector selector)
+			throws IOException, ClosedChannelException, SocketException {
+		while (true) {
+			int selectionCount = selector.select();
+			if (logger.isDebugEnabled())
+				logger.debug("SelectionCount: " + selectionCount);
+			if (selectionCount > 0) {
+				Set<SelectionKey> keys = selector.selectedKeys();
+				Iterator<SelectionKey> iterator = keys.iterator();
+				SocketChannel channel = null;
+				while (iterator.hasNext()) {
+					final SelectionKey key = iterator.next();
+					iterator.remove();
+					if (key.isAcceptable()) {
+						channel = server.accept();
+						channel.configureBlocking(false);
+						channel.register(selector, SelectionKey.OP_READ);
+						Socket socket = channel.socket();
+						setSocketOptions(socket);
+					}
+					else if (key.isReadable()) {
+						key.interestOps(key.interestOps() - key.readyOps());
+						if (key.attachment() == null) {
+							NioSocketReader reader = createSocketReader(key);
+							key.attach(reader);
+						}
+						this.threadPoolTaskScheduler.execute(new Runnable() {
+							public void run() {
+								doRead(key);
+								if (key.channel().isOpen()) {
+									key.interestOps(SelectionKey.OP_READ);
+									selector.wakeup();
+								}
+							}});
+					}
+					else {
+						logger.error("Unexpected key: " + key);
+					}
+				}
+			}			
+		}
+	}
+
+	/**
+	 * Creates an NioSocketReader, either directly,or 
+	 * from the supplied class if {@link MessageFormats#FORMAT_CUSTOM}
+	 * is used. 
+	 * @param key The selection key.
+	 * @return The NioSocketReader.
+	 */
+	private NioSocketReader createSocketReader(final SelectionKey key) {
+		NioSocketReader reader = null;
+		SocketChannel channel = (SocketChannel) key.channel();
+		if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
+			try {
+				reader = (NioSocketReader) customSocketReader.newInstance();
+				reader.setChannel(channel);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+		} else {
+			reader = new NioSocketReader(channel);
+		}
+		reader.setUsingDirectBuffers(usingDirectBuffers);
+		reader.setMessageFormat(messageFormat);
+		return reader;
+	}
+
+	/**
+	 * Obtains the {@link NetSocketReader} associated with the channel 
+	 * and calls its {@link NetSocketReader#assembledData}
+	 * method; if a message is fully assembled,  calls {@link #sendMessage(Message)} with the
+	 * mapped message.
+	 * 
+	 * @param channel
+	 */
+	private void doRead(SelectionKey key) {
+		NioSocketReader reader = (NioSocketReader) key.attachment();
+		try {
+			if (reader.assembleData()) {
+				Message<byte[]> message;
+					message = mapper.toMessage(reader);
+					if (message != null) {
+						sendMessage(message);
+					}
+			}
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+			try {
+				key.channel().close();
+			} catch (IOException e1) {
+				// TODO Auto-generated catch block
+				e1.printStackTrace();
+			}
+		}
+	}
+
+}
+ 

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

@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.net.Socket;
+
+/**
+ * Reads messages that are exactly 24 bytes long.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class CustomNetSocketReader extends NetSocketReader {
+
+	public CustomNetSocketReader() {
+		super(null);
+	}
+	
+	/**
+	 * @param socket
+	 */
+	public CustomNetSocketReader(Socket socket) {
+		super(socket);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat()
+	 */
+	@Override
+	protected boolean assembleDataCustomFormat() throws IOException {
+		byte[] buff = new byte[24];
+		read(buff);
+		assembledData = buff;
+		return true;
+	}
+	
+	
+
+}

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

@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+/**
+ * Reads messages that are exactly 24 bytes long.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class CustomNioSocketReader extends NioSocketReader {
+
+	private ByteBuffer buffer;
+
+	public CustomNioSocketReader() {
+		super(null);
+	}
+	
+	/**
+	 * @param socket
+	 */
+	public CustomNioSocketReader(SocketChannel channel) {
+		super(channel);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat()
+	 */
+	@Override
+	protected boolean assembleDataCustomFormat() throws IOException {
+		if (buffer == null) {
+			buffer = allocate(24);
+		}
+		readChannel(buffer);
+		if (buffer.hasRemaining()) {
+			return false;
+		}
+		assembledData = buffer.array();
+		buffer = null;
+		return true;
+	}
+	
+	
+
+}

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

@@ -0,0 +1,120 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+import javax.net.ServerSocketFactory;
+
+import org.junit.Test;
+
+/**
+ * @author Gary Russell
+ *
+ */
+public class NetSocketReaderTests {
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()},
+	 * using <length><message>.
+	 */
+	@Test
+	public void testReadLength() throws Exception {
+		int port = 23556;
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		Utils.testSendLength(port, null);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		NetSocketReader reader = new NetSocketReader(socket);
+		if (reader.assembleData()) {
+			assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+								 new String(reader.getAssembledData()));
+		}
+		else {
+			fail("Failed to assemble first message");
+		}
+		if (reader.assembleData()) {
+			assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+								 new String(reader.getAssembledData()));
+		}
+		else {
+			fail("Failed to assemble second message");
+		}
+	}
+	
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()},
+	 * using STX<message>ETX
+	 */
+	@Test
+	public void testReadStxEtx() throws Exception {
+		int port = 23557;
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		Utils.testSendStxEtx(port, null);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		NetSocketReader reader = new NetSocketReader(socket);
+		reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
+		if (reader.assembleData()) {
+			assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+								 new String(reader.getAssembledData()));
+		}
+		else {
+			fail("Failed to assemble first message");
+		}
+		if (reader.assembleData()) {
+			assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+								 new String(reader.getAssembledData()));
+		}
+		else {
+			fail("Failed to assemble second message");
+		}
+	}
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()},
+	 * using STX<message>ETX
+	 */
+	@Test
+	public void testReadCrLf() throws Exception {
+		int port = 23558;
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		Utils.testSendCrLf(port, null);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		NetSocketReader reader = new NetSocketReader(socket);
+		reader.setMessageFormat(MessageFormats.FORMAT_CRLF);
+		if (reader.assembleData()) {
+			assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+								 new String(reader.getAssembledData()));
+		}
+		else {
+			fail("Failed to assemble first message");
+		}
+		if (reader.assembleData()) {
+			assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+								 new String(reader.getAssembledData()));
+		}
+		else {
+			fail("Failed to assemble second message");
+		}
+	}
+
+}

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

@@ -0,0 +1,271 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.net.InetSocketAddress;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Test;
+
+/**
+ * @author Gary Russell
+ *
+ */
+public class NioSocketReaderTests {
+
+	private CountDownLatch latch = new CountDownLatch(1);
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
+	 */
+	@Test
+	public void testReadLength() throws Exception {
+		ServerSocketChannel server = ServerSocketChannel.open();
+		server.configureBlocking(false);
+		int port = 23456;
+		server.socket().bind(new InetSocketAddress(port));
+		final Selector selector = Selector.open();
+		server.register(selector, SelectionKey.OP_ACCEPT);
+		
+		// Fire up the sender.
+		Utils.testSendLength(port, latch);
+		
+		if(selector.select(10000) <= 0) {
+			fail("Socket failed to connect");
+		}
+		Set<SelectionKey> keys = selector.selectedKeys();
+		Iterator<SelectionKey> iterator = keys.iterator();
+		SocketChannel channel = null;
+		while (iterator.hasNext()) {
+			SelectionKey key = iterator.next();
+			iterator.remove();
+			if (key.isAcceptable()) {
+				channel = server.accept();
+				channel.configureBlocking(false);
+				channel.register(selector, SelectionKey.OP_READ);
+			}
+			else {
+				fail("Unexpected key: " + key);
+			}
+		}
+		NioSocketReader reader = new NioSocketReader(channel);
+		int count = 0;
+		while(selector.select(1000) > 0) {
+			keys = selector.selectedKeys();
+			iterator = keys.iterator();
+			while (iterator.hasNext()) {
+				SelectionKey key = iterator.next();
+				iterator.remove();
+				if (key.isReadable()) {
+					assertEquals(channel, key.channel());
+					if (reader.assembleData()) {
+						assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+						         new String(reader.getAssembledData()));
+						count++;
+					}
+					latch.countDown();
+				}
+				else {
+					fail("Unexpected key: " + key);
+				}
+			}
+		}
+		assertEquals("Did not receive data", 2, count);
+	}
+
+	@Test
+	public void testFragmented() throws Exception {
+		ServerSocketChannel server = ServerSocketChannel.open();
+		server.configureBlocking(false);
+		int port = 23457;
+		server.socket().bind(new InetSocketAddress(port));
+		final Selector selector = Selector.open();
+		server.register(selector, SelectionKey.OP_ACCEPT);
+		
+		// Fire up the sender.
+		Utils.testSendFragmented(port);
+		
+		if(selector.select(10000) <= 0) {
+			fail("Socket failed to connect");
+		}
+		Set<SelectionKey> keys = selector.selectedKeys();
+		Iterator<SelectionKey> iterator = keys.iterator();
+		SocketChannel channel = null;
+		while (iterator.hasNext()) {
+			SelectionKey key = iterator.next();
+			iterator.remove();
+			if (key.isAcceptable()) {
+				channel = server.accept();
+				channel.configureBlocking(false);
+				channel.register(selector, SelectionKey.OP_READ);
+			}
+			else {
+				fail("Unexpected key: " + key);
+			}
+		}
+		NioSocketReader reader = new NioSocketReader(channel);
+		boolean done = false;
+		while(selector.select(1000) > 0) {
+			keys = selector.selectedKeys();
+			iterator = keys.iterator();
+			while (iterator.hasNext()) {
+				SelectionKey key = iterator.next();
+				iterator.remove();
+				if (key.isReadable()) {
+					assertEquals(channel, key.channel());
+					if (reader.assembleData()) {
+						assertEquals("Data", "xx", 
+						         new String(reader.getAssembledData()));
+						done = true;
+					}
+					latch.countDown();
+				}
+				else {
+					fail("Unexpected key: " + key);
+				}
+			}
+		}
+		assertTrue("Did not receive data", done);
+	}
+	
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
+	 */
+	@Test
+	public void testReadStxEtx() throws Exception {
+		ServerSocketChannel server = ServerSocketChannel.open();
+		server.configureBlocking(false);
+		int port = 23458;
+		server.socket().bind(new InetSocketAddress(port));
+		final Selector selector = Selector.open();
+		server.register(selector, SelectionKey.OP_ACCEPT);
+		
+		// Fire up the sender.
+		Utils.testSendStxEtx(port, latch);
+		
+		if(selector.select(10000) <= 0) {
+			fail("Socket failed to connect");
+		}
+		Set<SelectionKey> keys = selector.selectedKeys();
+		Iterator<SelectionKey> iterator = keys.iterator();
+		SocketChannel channel = null;
+		while (iterator.hasNext()) {
+			SelectionKey key = iterator.next();
+			iterator.remove();
+			if (key.isAcceptable()) {
+				channel = server.accept();
+				channel.configureBlocking(false);
+				channel.register(selector, SelectionKey.OP_READ);
+			}
+			else {
+				fail("Unexpected key: " + key);
+			}
+		}
+		NioSocketReader reader = new NioSocketReader(channel);
+		reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
+		int count = 0;
+		while(selector.select(1000) > 0) {
+			keys = selector.selectedKeys();
+			iterator = keys.iterator();
+			while (iterator.hasNext()) {
+				SelectionKey key = iterator.next();
+				iterator.remove();
+				if (key.isReadable()) {
+					assertEquals(channel, key.channel());
+					if (reader.assembleData()) {
+						assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+						         new String(reader.getAssembledData()));
+						count++;
+					}
+					latch.countDown();
+				}
+				else {
+					fail("Unexpected key: " + key);
+				}
+			}
+		}
+		assertEquals("Did not receive data", 2, count);
+	}
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
+	 */
+	@Test
+	public void testReadCrLf() throws Exception {
+		ServerSocketChannel server = ServerSocketChannel.open();
+		server.configureBlocking(false);
+		int port = 23459;
+		server.socket().bind(new InetSocketAddress(port));
+		final Selector selector = Selector.open();
+		server.register(selector, SelectionKey.OP_ACCEPT);
+		
+		// Fire up the sender.
+		Utils.testSendCrLf(port, latch);
+		
+		if(selector.select(10000) <= 0) {
+			fail("Socket failed to connect");
+		}
+		Set<SelectionKey> keys = selector.selectedKeys();
+		Iterator<SelectionKey> iterator = keys.iterator();
+		SocketChannel channel = null;
+		while (iterator.hasNext()) {
+			SelectionKey key = iterator.next();
+			iterator.remove();
+			if (key.isAcceptable()) {
+				channel = server.accept();
+				channel.configureBlocking(false);
+				channel.register(selector, SelectionKey.OP_READ);
+			}
+			else {
+				fail("Unexpected key: " + key);
+			}
+		}
+		NioSocketReader reader = new NioSocketReader(channel);
+		reader.setMessageFormat(MessageFormats.FORMAT_CRLF);
+		int count = 0;
+		while(selector.select(1000) > 0) {
+			keys = selector.selectedKeys();
+			iterator = keys.iterator();
+			while (iterator.hasNext()) {
+				SelectionKey key = iterator.next();
+				iterator.remove();
+				if (key.isReadable()) {
+					assertEquals(channel, key.channel());
+					if (reader.assembleData()) {
+						assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, 
+						         new String(reader.getAssembledData()));
+						count++;
+					}
+					latch.countDown();
+				}
+				else {
+					fail("Unexpected key: " + key);
+				}
+			}
+		}
+		assertEquals("Did not receive data", 2, count);
+	}
+
+}

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

@@ -0,0 +1,98 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.ip.IpHeaders;
+
+/**
+ * @author Gary Russell
+ *
+ */
+public class SocketMessageMapperTests {
+
+	/**
+	 * 
+	 */
+	private static final String TEST_PAYLOAD = "abcdefghijkl";
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#toMessage(org.springframework.integration.ip.tcp.SocketReader)}.
+	 * Tests segmented reads into the payload and verifies reassembly.
+	 */
+	@Test
+	public void testToMessage() throws Exception {
+		SocketMessageMapper mapper = new SocketMessageMapper();
+		Message<byte[]> message = mapper.toMessage(new StubSocketReader());
+		assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload()));
+		assertEquals(InetAddress.getLocalHost().getHostName(), message
+				.getHeaders().get(IpHeaders.HOSTNAME));
+		assertEquals(InetAddress.getLocalHost().getHostAddress(), message
+				.getHeaders().get(IpHeaders.IP_ADDRESS));
+	}
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#fromMessage(org.springframework.integration.core.Message)}.
+	 */
+	@Test
+	@Ignore
+	public void testFromMessage() {
+		fail("Not yet implemented");
+	}
+
+
+	private class StubSocketReader implements SocketReader {
+	
+		/* (non-Javadoc)
+		 * @see org.springframework.integration.ip.tcp.SocketReader#getAddress()
+		 */
+		public InetAddress getAddress() {
+			try {
+				return InetAddress.getLocalHost();
+			} catch (UnknownHostException e) {
+				fail("Unexpected Exception: " + e.getMessage());
+			}
+			return null;
+		}
+	
+		public byte[] getAssembledData() {
+			return TEST_PAYLOAD.getBytes();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.integration.ip.tcp.SocketReader#assembleData()
+		 */
+		public boolean assembleData() {
+			return false;
+		}
+
+		
+	}
+	
+	private class StubSocketWriter implements SocketWriter {
+	
+		
+	}
+	
+}

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

@@ -0,0 +1,142 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Test;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * @author Gary Russell
+ *
+ */
+public class TcpReceivingChannelAdapterTests {
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter#run()}.
+	 */
+	@Test
+	public void testNet() throws Exception {
+		QueueChannel channel = new QueueChannel(2);
+		int port = 12345;
+		AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port);
+		adapter.setOutputChannel(channel);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.start();
+		Thread.sleep(2000); // wait for server to start listening
+		Utils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice
+		Thread.sleep(2000); // wait for asynch processing
+		Message<?> message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+		message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+	}
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter#run()}.
+	 * Verifies operation of custom message formats.
+	 */
+	@Test
+	public void testNetCustom() throws Exception {
+		QueueChannel channel = new QueueChannel(2);
+		int port = 12346;
+		AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port);
+		adapter.setOutputChannel(channel);
+		adapter.setCustomSocketReader("org.springframework.integration.ip.tcp.CustomNetSocketReader");
+		adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.start();
+		Thread.sleep(2000); // wait for server to start listening
+		Utils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice
+		Thread.sleep(4000); // wait for asynch processing
+		Message<?> message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", 
+				new String((byte[])message.getPayload()));
+		message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", 
+				new String((byte[])message.getPayload()));
+	}
+
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter#run()}.
+	 */
+	@Test
+	public void testNio() throws Exception {
+		QueueChannel channel = new QueueChannel(2);
+		int port = 12355;
+		TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port);
+		adapter.setOutputChannel(channel);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.start();
+		Thread.sleep(2000); // wait for server to start listening
+		Utils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice
+		Thread.sleep(2000); // wait for asynch processing
+		Message<?> message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+		message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, 
+				new String((byte[])message.getPayload()));
+	}
+
+	/**
+	 * Test method for {@link org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter#run()}.
+	 * Verifies operation of custom message formats.	 */
+	@Test
+	public void testNioCustom() throws Exception {
+		QueueChannel channel = new QueueChannel(2);
+		int port = 12356;
+		TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port);
+		adapter.setOutputChannel(channel);
+		adapter.setCustomSocketReader("org.springframework.integration.ip.tcp.CustomNioSocketReader");
+		adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.start();
+		Thread.sleep(2000); // wait for server to start listening
+		Utils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice
+		Thread.sleep(4000); // wait for asynch processing
+		Message<?> message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", 
+				new String((byte[])message.getPayload()));
+		message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", 
+				new String((byte[])message.getPayload()));
+	}
+
+}

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java

@@ -0,0 +1,167 @@
+/*
+ * 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;
+
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * TCP/IP Test utilities.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class Utils {
+
+	public static final String TEST_STRING = "TestMessage";
+
+	/**
+	 * Sends a message in two chunks with a preceding length. Two such messages are sent.
+	 * @param latch If not null, await until counted down before sending second chunk.
+	 */
+	public static void testSendLength(final int port, final CountDownLatch latch) {
+		Thread thread = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = new Socket(InetAddress.getByName("localhost"), port);
+					for (int i = 0; i < 2; i++) {
+						byte[] len = new byte[4];
+						ByteBuffer.wrap(len).putInt(TEST_STRING.length() * 2);
+						socket.getOutputStream().write(len);
+						socket.getOutputStream().write(TEST_STRING.getBytes());
+						System.out.println(i + " Wrote first part");
+						if (latch != null) {
+							latch.await();
+						}
+						Thread.sleep(500);
+						// send the second chunk
+						socket.getOutputStream().write(TEST_STRING.getBytes());
+						System.out.println(i + " Wrote second part");
+					}
+					Thread.sleep(1000000000L); // wait forever, but we're a daemon
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		thread.setDaemon(true);
+		thread.start();
+	}
+
+	/**
+	 * Test for reassembly of completely fragmented message; sends
+	 * 6 bytes 500ms apart.
+	 * @param os
+	 * @param b
+	 * @throws Exception
+	 */
+	public static void testSendFragmented(final int port) {
+		Thread thread = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = new Socket(InetAddress.getByName("localhost"), port);
+					OutputStream os = socket.getOutputStream();
+					writeByte(os, 0);
+					writeByte(os, 0);
+					writeByte(os, 0);
+					writeByte(os, 2);
+					writeByte(os, 'x');
+					writeByte(os, 'x');
+					Thread.sleep(1000000000L); // wait forever, but we're a daemon
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		thread.setDaemon(true);
+		thread.start();
+	}
+
+	private static void writeByte(OutputStream os, int b) throws Exception {
+		os.write(b);
+		System.out.printf("Wrote 0x%x\n", b);
+		Thread.sleep(500);
+	}
+	
+	/**
+	 * Sends a STX/ETX message in two chunks. Two such messages are sent.
+	 * @param latch If not null, await until counted down before sending second chunk.
+	 */
+	public static void testSendStxEtx(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();
+					for (int i = 0; i < 2; i++) {
+						writeByte(outputStream, 0x02);
+						outputStream.write(TEST_STRING.getBytes());
+						System.out.println(i + " Wrote first part");
+						if (latch != null) {
+							latch.await();
+						}
+						Thread.sleep(500);
+						// send the second chunk
+						outputStream.write(TEST_STRING.getBytes());
+						System.out.println(i + " Wrote second part");
+						writeByte(outputStream, 0x03);
+					}
+					Thread.sleep(1000000000L); // wait forever, but we're a daemon
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		thread.setDaemon(true);
+		thread.start();
+	}
+	
+	/**
+	 * Sends a message +CRLF in two chunks. Two such messages are sent.
+	 * @param latch If not null, await until counted down before sending second chunk.
+	 */
+	public static void testSendCrLf(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();
+					for (int i = 0; i < 2; i++) {
+						outputStream.write(TEST_STRING.getBytes());
+						System.out.println(i + " Wrote first part");
+						if (latch != null) {
+							latch.await();
+						}
+						Thread.sleep(500);
+						// send the second chunk
+						outputStream.write(TEST_STRING.getBytes());
+						System.out.println(i + " Wrote second part");
+						writeByte(outputStream, '\r');
+						writeByte(outputStream, '\n');
+					}
+					Thread.sleep(1000000000L); // wait forever, but we're a daemon
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		thread.setDaemon(true);
+		thread.start();
+	}
+}

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.ip;
+package org.springframework.integration.ip.udp;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.ip;
+package org.springframework.integration.ip.udp;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.ip;
+package org.springframework.integration.ip.udp;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -32,6 +32,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.channel.ChannelResolver;
 import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.ip.StdOutCatcher;
 import org.springframework.integration.message.StringMessage;
 
 /**

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.ip;
+package org.springframework.integration.ip.udp;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -30,15 +30,15 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.channel.ChannelResolver;
 import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.ip.StdOutCatcher;
 import org.springframework.integration.message.StringMessage;
 
 /**
  * Sends and receives a simple message through to the Udp channel adapters.
  * If run as a JUnit just sends one message and terminates (see console).
- * TODO: Use a custom output stream and catch output to verify.
  * 
  * If run from main(),  
- * hangs around for a couple of minutes to allow console interaction (enter a message on the
+ * hangs around for a couple of minutes to allow console interaction - enter a message on the
  * console and you should see it go through the outbound context, over UDP, and 
  * received in the other context (and written back to the console).
  *  

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml

@@ -23,7 +23,7 @@
 
 	<beans:bean id="stdoutCatcher" class = "org.springframework.integration.ip.StdOutCatcher"/>
 
-	<stream:stderr-channel-adapter id="stdout" channel="udpToStdOutChannel" append-newline="true"/> 
+	<stream:stdout-channel-adapter id="stdout" channel="udpToStdOutChannel" append-newline="true"/> 
 
 	<beans:bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
 		<beans:property name="daemon" value="true" />