INT-1877 Support Byte, Short, and Int Length Headers

docs/src/reference/docbook/ip.xml

@@ -239,11 +239,17 @@
      </para>
      <para> 
       The <classname>ByteArrayLengthHeaderSerializer</classname>,
-      converts a byte array to a stream of bytes preceded by a 4 byte binary
-      length in network byte order. This a very efficient deserializer 
+      converts a byte array to a stream of bytes preceded by a binary
+      length in network byte order (big endian). This a very efficient deserializer 
       because it does not have to parse every byte looking for a termination
       character sequence. It can also be used for payloads containing binary data;
-      the above serializers only support text in the payload.
+      the above serializers only support text in the payload. The default size of 
+      the length header is 4 bytes (Integer), allowing for messages up to 2**31-1
+      bytes. However, the length header can be a single byte (unsigned) for
+      messages up to 255 bytes, or an unsigned short (2 bytes) for messages up to
+      2**16 bytes. If you need any other format for the header, you can subclass
+      this class and provide implementations for the readHeader and writeHeader
+      methods. The absolute maximum data size supported is 2**31-1 bytes.
      </para>
      <para> 
       The <classname>ByteArrayRawSerializer</classname>,
@@ -288,7 +294,10 @@
      <para>
       To implement a custom (de)serializer pair, implement the
       <classname>org.springframework.core.serializer.Deserializer</classname> and
-      <classname>org.springframework.core.serializer.Serializer</classname> interfaces. If you do not wish to use
+      <classname>org.springframework.core.serializer.Serializer</classname> interfaces. 
+     </para>
+     <para> 
+      If you do not wish to use
       the default (de)serializer (<classname>ByteArrayCrLfSerializer</classname>), you must supply
       <classname>serializer</classname> and
       <classname>deserializer</classname> attributes on the connection factory (example below).
@@ -303,7 +312,7 @@
   <ip:tcp-connection-factory id="server"
     type="server"
     port="1234"
-    deserializer="JavaDeserializer"
+    deserializer="javaDeserializer"
     serializer="javaSerializer"
   />]]></programlisting>
       A server connection factory that uses <classname>java.net.Socket</classname>

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

@@ -26,34 +26,75 @@ 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.
+ * a binary length (network byte order, not included in resulting byte[]). 
+ * 
+ * Writes a byte[] to an OutputStream after a binary length.
  * The length field contains the length of data following the length
- * field.
- * (network byte order).
+ * field. (network byte order).
+ * 
+ * The default length field is a 4 byte signed integer. During deserialization,
+ * negative values will be rejected.
+ * Other options are an unsigned byte, and unsigned short.
+ * 
+ * For other header formats, override {@link #readHeader(InputStream)} and
+ * {@link #writeHeader(OutputStream, int)}.
  * 
  * @author Gary Russell
  * @since 2.0
  */
 public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer {
-
+	
+	
+	/**
+	 * Default length-header field, allows for data up to 2**31-1 bytes.
+	 */
+	public static final int HEADER_SIZE_INT = 4; // default
+	
+	/**
+	 * A single unsigned byte, for data up to 255 bytes.
+	 */
+	public static final int HEADER_SIZE_UNSIGNED_BYTE = 1;
+	
+	/**
+	 * An unsigned short, for data up to 2**16 bytes.
+	 */
+	public static final int HEADER_SIZE_UNSIGNED_SHORT = 2;
+	
+	private final int headerSize;
+	
 	private Log logger = LogFactory.getLog(this.getClass());
 	
 	/**
-	 * Reads a 4 byte length from the stream and then reads that length
+	 * Constructs the serializer using {@link #HEADER_SIZE_INT}
+	 */
+	public ByteArrayLengthHeaderSerializer() {
+		this(HEADER_SIZE_INT);
+	}
+	
+	/**
+	 * Constructs the serializer using the supplied header size.
+	 * Valid header sizes are {@link #HEADER_SIZE_INT} (default),
+	 * {@link #HEADER_SIZE_UNSIGNED_BYTE} and {@link #HEADER_SIZE_UNSIGNED_SHORT}
+	 * @param headerSize
+	 */
+	public ByteArrayLengthHeaderSerializer(int headerSize) {
+		if (headerSize != HEADER_SIZE_INT &&
+			headerSize != HEADER_SIZE_UNSIGNED_BYTE &&
+			headerSize != HEADER_SIZE_UNSIGNED_SHORT) {
+			throw new IllegalArgumentException("Illegal header size:" + headerSize);
+		}
+		this.headerSize = headerSize;
+	}
+	
+	/**
+	 * Reads the header from the stream and then reads the provided 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();
+		int messageLength = this.readHeader(inputStream);
 		if (logger.isDebugEnabled()) {
 			logger.debug("Message length is " + messageLength);
 		}	
@@ -71,9 +112,7 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer
 	 * 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());
+		this.writeHeader(outputStream, bytes.length);
 		outputStream.write(bytes);
 		outputStream.flush();
 	}
@@ -110,4 +149,72 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer
 		return 0;
 	}
 
+	/**
+	 * Writes the header, according to the header format.
+	 * @param outputStream
+	 * @param length
+	 * @throws IOException 
+	 */
+	protected void writeHeader(OutputStream outputStream, int length) throws IOException {
+		ByteBuffer lengthPart = ByteBuffer.allocate(this.headerSize);
+		switch (this.headerSize) {
+		case HEADER_SIZE_INT:
+			lengthPart.putInt(length);			
+			break;
+		case HEADER_SIZE_UNSIGNED_BYTE:
+			if (length > 0xff) {
+				throw new IllegalArgumentException("Length header:"
+						+ headerSize
+						+ " too short to accommodate message length:" + length);
+			}
+			lengthPart.put((byte) length);
+			break;
+		case HEADER_SIZE_UNSIGNED_SHORT:
+			if (length > 0xffff) {
+				throw new IllegalArgumentException("Length header:"
+						+ headerSize
+						+ " too short to accommodate message length:" + length);
+			}
+			lengthPart.putShort((short) length);
+			break;
+		default:
+			throw new IllegalArgumentException("Bad header size:" + headerSize);			
+		}
+		outputStream.write(lengthPart.array());
+	}
+
+	/**
+	 * Reads the header and returns the length of the data part.
+	 * @param inputStream
+	 * @return The length of the data part
+	 * @throws IOException, {@link SoftEndOfStreamException} if socket closes
+	 * before any length data read.
+	 */
+	protected int readHeader(InputStream inputStream) throws IOException {
+		byte[] lengthPart = new byte[this.headerSize];
+		int status = read(inputStream, lengthPart, true);
+		if (status < 0) {
+			throw new SoftEndOfStreamException("Stream closed between payloads");
+		}
+		int messageLength;
+		switch (this.headerSize) {
+		case HEADER_SIZE_INT:
+			messageLength = ByteBuffer.wrap(lengthPart).getInt();
+			if (messageLength < 0) {
+				throw new IllegalArgumentException("Length header:"
+						+ messageLength 
+						+ " is negative");
+			}
+			break;
+		case HEADER_SIZE_UNSIGNED_BYTE:
+			messageLength = ByteBuffer.wrap(lengthPart).get() & 0xff;
+			break;
+		case HEADER_SIZE_UNSIGNED_SHORT:
+			messageLength = ByteBuffer.wrap(lengthPart).getShort() & 0xffff;
+			break;
+		default:
+			throw new IllegalArgumentException("Bad header size:" + headerSize);			
+		}
+		return messageLength;
+	}
 }

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

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2002-2011 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.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.util.Arrays;
+
+import org.junit.Before;
+import org.junit.Test;
+
+
+
+/**
+ * @author Gary Russell
+ * @since 2.0.4
+ *
+ */
+public class LenghtHeaderSerializationTests {
+
+	private static final String TEST = "Test";
+	private String test255;
+	private String testFFFF;
+	
+	@Before
+	public void setup() {
+		char[] chars = new char[255]; 
+		Arrays.fill(chars, 'x');
+		test255 = new String(chars);
+		chars = new char[0xffff];
+		Arrays.fill(chars, 'x');
+		testFFFF = new String(chars);
+	}
+
+	@Test
+	public void testInt() throws Exception {
+		AbstractByteArraySerializer serializer = new ByteArrayLengthHeaderSerializer();
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		serializer.serialize(TEST.getBytes(), bos);
+		byte[] bytes = bos.toByteArray();
+		assertEquals(0, bytes[0]);
+		assertEquals(0, bytes[1]);
+		assertEquals(0, bytes[2]);
+		assertEquals(TEST.length(), bytes[3]);
+		ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+		bytes = serializer.deserialize(bis);
+		assertEquals(TEST, new String(bytes));
+		bytes[0] = -1;
+		bis = new ByteArrayInputStream(bytes);
+		try {
+			bytes = serializer.deserialize(bis);
+			fail("Expected negative length");
+		} catch (IllegalArgumentException e) { }
+	}
+
+	@Test
+	public void testByte() throws Exception {
+		AbstractByteArraySerializer serializer = new ByteArrayLengthHeaderSerializer(
+				ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_BYTE);
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		serializer.serialize(test255.getBytes(), bos);
+		byte[] bytes = bos.toByteArray();
+		assertEquals(test255.length(), bytes[0] & 0xff);
+		ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+		bytes = serializer.deserialize(bis);
+		assertEquals(test255, new String(bytes));
+		test255 += "x";
+		try {
+			serializer.serialize(test255.getBytes(), bos);
+			fail("Expected overflow");
+		} catch (IllegalArgumentException e) { }
+	}
+
+	@Test
+	public void testShort1() throws Exception {
+		AbstractByteArraySerializer serializer = new ByteArrayLengthHeaderSerializer(
+				ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_SHORT);
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		serializer.serialize(test255.getBytes(), bos);
+		byte[] bytes = bos.toByteArray();
+		assertEquals(0, bytes[0]);		
+		assertEquals(test255.length(), bytes[1] & 0xff);
+		ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+		bytes = serializer.deserialize(bis);
+		assertEquals(test255, new String(bytes));
+	}
+
+	@Test
+	public void testShort2() throws Exception {
+		AbstractByteArraySerializer serializer = new ByteArrayLengthHeaderSerializer(
+				ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_SHORT);
+		serializer.setMaxMessageSize(0x10000);
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		serializer.serialize(testFFFF.getBytes(), bos);
+		byte[] bytes = bos.toByteArray();
+		assertEquals(0xff, bytes[0] & 0xff);
+		assertEquals(0xff, bytes[1] & 0xff);
+		ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+		bytes = serializer.deserialize(bis);
+		assertEquals(testFFFF, new String(bytes));
+		testFFFF += "x";
+		try {
+			serializer.serialize(testFFFF.getBytes(), bos);
+			fail("Expected overflow");
+		} catch (IllegalArgumentException e) { }
+	}
+
+	@Test
+	public void testBad() throws Exception {
+		try {
+			new ByteArrayLengthHeaderSerializer(23);
+			fail("Expected illegal argument exception");
+		} catch (IllegalArgumentException e) { }
+
+	}
+}