diff --git a/docs/src/reference/docbook/ip.xml b/docs/src/reference/docbook/ip.xml index 1f724b9d6a..8a8dbf856d 100644 --- a/docs/src/reference/docbook/ip.xml +++ b/docs/src/reference/docbook/ip.xml @@ -239,11 +239,17 @@ The ByteArrayLengthHeaderSerializer, - 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. The ByteArrayRawSerializer, @@ -288,7 +294,10 @@ To implement a custom (de)serializer pair, implement the org.springframework.core.serializer.Deserializer and - org.springframework.core.serializer.Serializer interfaces. If you do not wish to use + org.springframework.core.serializer.Serializer interfaces. + + + If you do not wish to use the default (de)serializer (ByteArrayCrLfSerializer), you must supply serializer and deserializer attributes on the connection factory (example below). @@ -303,7 +312,7 @@ ]]> A server connection factory that uses java.net.Socket diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java index d76067d8f2..58afe2f35e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java +++ b/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; + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/LenghtHeaderSerializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/LenghtHeaderSerializationTests.java new file mode 100644 index 0000000000..bae0989f6b --- /dev/null +++ b/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) { } + + } +}