INT-1486 polish; rename byte array 'serializers' and all test usage

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

@@ -0,0 +1,68 @@
+/*
+ * 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.serializer;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.commons.serializer.Deserializer;
+import org.springframework.commons.serializer.Serializer;
+
+/**
+ * Base class for (de)serializers that provide a mechanism to 
+ * reconstruct a byte array from an arbitrary stream.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public abstract class AbstractByteArraySerializer implements
+		Serializer<byte[]>, 
+		Deserializer<byte[]> {
+
+	protected int maxMessageSize = 2048;
+	
+	protected Log logger = LogFactory.getLog(this.getClass());
+
+	/**
+	 * The maximum supported message size for this serializer.
+	 * Default 2048.
+	 * @return The max message size.
+	 */
+	public int getMaxMessageSize() {
+		return maxMessageSize;
+	}
+
+	/**
+	 * The maximum supported message size for this serializer.
+	 * Default 2048.
+	 * @param maxMessageSize The max message size.
+	 */
+	public void setMaxMessageSize(int maxMessageSize) {
+		this.maxMessageSize = maxMessageSize;
+	}
+
+	protected void checkClosure(int bite) throws IOException {
+		if (bite < 0) {
+			logger.debug("Socket closed");				
+			throw new IOException("Socket closed");
+		}
+	}
+
+}

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

@@ -0,0 +1,75 @@
+/*
+ * 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.serializer;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * Reads data in an InputStream to a byte[]; data must be terminated by \r\n
+ * (not included in resulting byte[]). 
+ * Writes a byte[] to an OutputStream and adds \r\n.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer {
+
+	/**
+	 * Reads the data in the inputstream to a byte[]. Data must be terminated
+	 * by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream
+	 * is closed immediately after the \r\n (i.e. no data is in the process of
+	 * being read). 
+	 */
+	public byte[] deserialize(InputStream inputStream) throws IOException {
+		byte[] buffer = new byte[this.maxMessageSize];
+		int n = 0;
+		int bite;
+		if (logger.isDebugEnabled()) 
+			logger.debug("Available to read:" + inputStream.available());
+		while (true) {
+			bite = inputStream.read();
+//			logger.debug("Read:" + (char) bite);
+			if (bite < 0 && n == 0) {
+				throw new SoftEndOfStreamException("Stream closed between payloads");
+			}
+			checkClosure(bite);
+			if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
+				break;
+			buffer[n++] = (byte) bite;
+			if (n >= this.maxMessageSize) {
+				throw new IOException("CRLF not found before max message length: "
+						+ this.maxMessageSize);
+			}
+		};
+		byte[] assembledData = new byte[n-1];
+		System.arraycopy(buffer, 0, assembledData, 0, n-1);
+		return assembledData;
+	}
+
+	/**
+	 * Writes the byte[] to the stream and appends \r\n.
+	 */
+	public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
+		outputStream.write(bytes);
+		outputStream.write('\r');
+		outputStream.write('\n');
+		outputStream.flush();
+	}
+
+}

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

@@ -0,0 +1,113 @@
+/*
+ * 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.serializer;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Reads data in an InputStream to a byte[]; data must be preceded by
+ * a 4 byte binary length (network byte order, 
+ * not included in resulting byte[]). 
+ * Writes a byte[] to an OutputStream after a 4 byte binary length.
+ * The length field contains the length of data following the length
+ * field.
+ * (network byte order).
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer {
+
+	private Log logger = LogFactory.getLog(this.getClass());
+	
+	/**
+	 * Reads a 4 byte length from the stream and then reads that length
+	 * from the stream and returns the data in a byte[]. Throws an
+	 * IOException if the length field exceeds the maxMessageSize.
+	 * Throws a {@link SoftEndOfStreamException} if the stream
+	 * is closed between messages.  
+	 */
+	public byte[] deserialize(InputStream inputStream) throws IOException {
+		byte[] lengthPart = new byte[4];
+		int status = read(inputStream, lengthPart, true);
+		if (status < 0) {
+			throw new SoftEndOfStreamException("Stream closed between payloads");
+		}
+		int messageLength = ByteBuffer.wrap(lengthPart).getInt();
+		if (logger.isDebugEnabled()) {
+			logger.debug("Message length is " + messageLength);
+		}	
+		if (messageLength > this.maxMessageSize) {
+			throw new IOException("Message length " + messageLength + 
+					" exceeds max message length: " + this.maxMessageSize);
+		}
+		byte[] messagePart = new byte[messageLength];
+		read(inputStream, messagePart, false);
+		return messagePart;
+	}
+
+	/**
+	 * Writes the byte[] to the output stream, preceded by a 4 byte
+	 * length in network byte order (big endian).
+	 */
+	public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
+		ByteBuffer lengthPart = ByteBuffer.allocate(4);
+		lengthPart.putInt(bytes.length);
+		outputStream.write(lengthPart.array());
+		outputStream.write(bytes);
+		outputStream.flush();
+	}
+
+	/**
+	 * 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 int read(InputStream inputStream, byte[] buffer, boolean header) 
+			throws IOException {
+		int lengthRead = 0;
+		int needed = buffer.length;
+		while (lengthRead < needed) {
+			int len;
+			len = inputStream.read(buffer, lengthRead,
+					needed - lengthRead);
+			if (len < 0 && header && lengthRead == 0) {
+				return len;
+			}
+			if (len < 0) {
+				throw new IOException("Stream closed after " + lengthRead + " of " + needed);
+			}
+			lengthRead += len;
+			if (logger.isDebugEnabled()) {
+				logger.debug("Read " + len + " bytes, buffer is now at " + 
+							 lengthRead + " of " +
+							 needed);
+			}
+		}
+		return 0;
+	}
+
+}

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

@@ -0,0 +1,80 @@
+/*
+ * 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.serializer;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.springframework.integration.mapping.MessageMappingException;
+
+/**
+ * Reads data in an InputStream to a byte[]; data must be prefixed by <stx> and
+ * terminated by <etx> (not included in resulting byte[]). 
+ * Writes a byte[] to an OutputStream prefixed by <stx> terminated by <etx>
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer {
+
+	public static final int STX = 0x02;
+	
+	public static final int ETX = 0x03;
+
+	/**
+	 * Reads the data in the inputstream to a byte[]. Data must be prefixed 
+	 * with an ASCII STX character, and terminated with an ASCII ETX character.
+	 * Throws a {@link SoftEndOfStreamException} if the stream
+	 * is closed immediately before the STX (i.e. no data is in the process of
+	 * being read). 
+	 *  
+	 */
+	public byte[] deserialize(InputStream inputStream) throws IOException {
+		int bite = inputStream.read();
+		if (bite < 0) {
+			throw new SoftEndOfStreamException("Stream closed between payloads");
+		}
+		if (bite != STX)
+			throw new MessageMappingException("Expected STX to begin message");
+		byte[] buffer = new byte[this.maxMessageSize];
+		int n = 0;
+		while ((bite = inputStream.read()) != ETX) {
+			checkClosure(bite);
+			buffer[n++] = (byte) bite;
+			if (n >= this.maxMessageSize) {
+				throw new IOException("ETX not found before max message length: "
+						+ this.maxMessageSize);
+			}
+		}
+		byte[] assembledData = new byte[n];
+		System.arraycopy(buffer, 0, assembledData, 0, n);
+		return assembledData;
+	}
+
+	/**
+	 * Writes the byte[] to the stream, prefixed by an ASCII STX character and
+	 * terminated with an ASCII ETX character.
+	 */
+	public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
+		outputStream.write(STX);
+		outputStream.write(bytes);
+		outputStream.write(ETX);
+		outputStream.flush();
+	}
+
+}

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

@@ -0,0 +1,41 @@
+/*
+ * 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.serializer;
+
+import java.io.IOException;
+
+/**
+ * Used to communicate that a stream has closed, but between logical
+ * messages.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class SoftEndOfStreamException extends IOException {
+
+	private static final long serialVersionUID = 7309907445617226978L;
+
+	public SoftEndOfStreamException() {
+		super();
+	}
+
+	public SoftEndOfStreamException(String message) {
+		super(message);
+	}
+
+}

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

@@ -0,0 +1,6 @@
+/**
+ * Byte array (de)serializers for putting some protocol on the
+ * wire so that incoming messages can be constructed from stream data.
+ */
+package org.springframework.integration.ip.tcp.serializer;
+

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

@@ -0,0 +1,219 @@
+/*
+ * 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.serializer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+
+import javax.net.ServerSocketFactory;
+
+import org.junit.Test;
+
+import org.springframework.commons.serializer.DefaultDeserializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
+import org.springframework.integration.ip.util.SocketUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class DeserializationTests {
+
+	@Test
+	public void testReadLength() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendLength(port, null);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
+		byte[] out = serializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+								 new String(out));
+		out = serializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				 new String(out));
+		server.close();
+	}
+	
+	@Test
+	public void testReadStxEtx() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendStxEtx(port, null);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
+		byte[] out = serializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+								 new String(out));
+		out = serializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				 new String(out));
+		server.close();
+	}
+
+	@Test
+	public void testReadCrLf() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendCrLf(port, null);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
+		byte[] out = serializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+								 new String(out));
+		out = serializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, 
+				 new String(out));
+		server.close();
+	}
+
+	@Test
+	public void testReadSerialized() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendSerialized(port);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		DefaultDeserializer deserializer = new DefaultDeserializer();
+		Object out = deserializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING, out);
+		out = deserializer.deserialize(socket.getInputStream());
+		assertEquals("Data", SocketUtils.TEST_STRING, out);
+		server.close();
+	}
+
+	@Test
+	public void testReadLengthOverflow() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendLengthOverflow(port);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
+		try {
+			serializer.deserialize(socket.getInputStream());
+	    	fail("Expected message length exceeded exception");
+		} catch (IOException e) {
+			if (!e.getMessage().startsWith("Message length")) {
+				e.printStackTrace();
+				fail("Unexpected IO Error:" + e.getMessage());
+			}
+		}
+		server.close();
+	}
+
+	@Test
+	public void testReadStxEtxTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendStxEtxOverflow(port);
+		Socket socket = server.accept();
+		socket.setSoTimeout(500);
+		ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
+		try {
+			serializer.deserialize(socket.getInputStream());
+	    	fail("Expected timeout exception");
+		} catch (IOException e) {
+			if (!e.getMessage().startsWith("Read timed out")) {
+				e.printStackTrace();
+				fail("Unexpected IO Error:" + e.getMessage());
+			}
+		}
+		server.close();
+	}
+
+	@Test
+	public void testReadStxEtxOverflow() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendStxEtxOverflow(port);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
+		serializer.setMaxMessageSize(1024);
+		try {
+			serializer.deserialize(socket.getInputStream());
+	    	fail("Expected message length exceeded exception");
+		} catch (IOException e) {
+			if (!e.getMessage().startsWith("ETX not found")) {
+				e.printStackTrace();
+				fail("Unexpected IO Error:" + e.getMessage());
+			}
+		}
+		server.close();
+	}
+
+	@Test
+	public void testReadCrLfTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendCrLfOverflow(port);
+		Socket socket = server.accept();
+		socket.setSoTimeout(500);
+		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
+		try {
+			serializer.deserialize(socket.getInputStream());
+	    	fail("Expected timout exception");
+		} catch (IOException e) {
+			if (!e.getMessage().startsWith("Read timed out")) {
+				e.printStackTrace();
+				fail("Unexpected IO Error:" + e.getMessage());
+			}
+		}
+		server.close();
+	}
+
+	@Test
+	public void testReadCrLfOverflow() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		SocketUtils.testSendCrLfOverflow(port);
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
+		serializer.setMaxMessageSize(1024);
+		try {
+			serializer.deserialize(socket.getInputStream());
+	    	fail("Expected message length exceeded exception");
+		} catch (IOException e) {
+			if (!e.getMessage().startsWith("CRLF not found")) {
+				e.printStackTrace();
+				fail("Unexpected IO Error:" + e.getMessage());
+			}
+		}
+		server.close();
+	}
+
+}

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

@@ -0,0 +1,185 @@
+/*
+ * 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.serializer;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+
+import org.junit.Test;
+
+import org.springframework.commons.serializer.DefaultSerializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
+import org.springframework.integration.ip.util.SocketUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class SerializationTests {
+
+	@Test
+	public void testWriteLengthHeader() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		final String testString = "abcdef";
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		Thread t = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+					ByteBuffer buffer = ByteBuffer.allocate(testString.length());
+					buffer.put(testString.getBytes());
+					ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
+					serializer.serialize(buffer.array(), socket.getOutputStream());
+					Thread.sleep(1000000000L);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		t.setDaemon(true);
+		t.start();
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		InputStream is = socket.getInputStream();
+		byte[] buff = new byte[testString.length() + 4];
+		readFully(is, buff);
+		ByteBuffer buffer = ByteBuffer.wrap(buff);
+		assertEquals(testString.length(), buffer.getInt());
+		assertEquals(testString, new String(buff, 4, testString.length()));
+		server.close();
+	}
+
+	@Test
+	public void testWriteStxEtx() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		final String testString = "abcdef";
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		Thread t = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+					ByteBuffer buffer = ByteBuffer.allocate(testString.length());
+					buffer.put(testString.getBytes());
+					ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
+					serializer.serialize(buffer.array(), socket.getOutputStream());
+					Thread.sleep(1000000000L);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		t.setDaemon(true);
+		t.start();
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		InputStream is = socket.getInputStream();
+		byte[] buff = new byte[testString.length() + 2];
+		readFully(is, buff);
+		assertEquals(ByteArrayStxEtxSerializer.STX, buff[0]);
+		assertEquals(testString, new String(buff, 1, testString.length()));
+		assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
+		server.close();
+	}
+
+	@Test
+	public void testWriteCrLf() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		final String testString = "abcdef";
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		Thread t = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+					ByteBuffer buffer = ByteBuffer.allocate(testString.length());
+					buffer.put(testString.getBytes());
+					ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
+					serializer.serialize(buffer.array(), socket.getOutputStream());
+					Thread.sleep(1000000000L);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		t.setDaemon(true);
+		t.start();
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		InputStream is = socket.getInputStream();
+		byte[] buff = new byte[testString.length() + 2];
+		readFully(is, buff);
+		assertEquals(testString, new String(buff, 0, testString.length()));
+		assertEquals('\r', buff[testString.length()]);
+		assertEquals('\n', buff[testString.length() + 1]);
+		server.close();
+	}
+	
+	@Test
+	public void testWriteSerialized() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		final String testString = "abcdef";
+		ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+		server.setSoTimeout(10000);
+		Thread t = new Thread(new Runnable() {
+			public void run() {
+				try {
+					Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+					DefaultSerializer serializer = new DefaultSerializer();
+					serializer.serialize(testString, socket.getOutputStream());
+					serializer.serialize(testString, socket.getOutputStream());
+					Thread.sleep(1000000000L);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		t.setDaemon(true);
+		t.start();
+		Socket socket = server.accept();
+		socket.setSoTimeout(5000);
+		InputStream is = socket.getInputStream();
+		ObjectInputStream ois = new ObjectInputStream(is);
+		assertEquals(testString, ois.readObject());
+		ois = new ObjectInputStream(is);
+		assertEquals(testString, ois.readObject());
+		server.close();
+	}
+	
+	/**
+	 * @param is
+	 * @param buff
+	 */
+	private void readFully(InputStream is, byte[] buff) throws IOException {
+		for (int i = 0; i < buff.length; i++) {
+			buff[i] = (byte) is.read();
+		}
+	}
+
+}