INT-4029: TCP: Add Buffer Pooling to Deserializers

JIRA: https://jira.spring.io/browse/INT-4029

Support the use of buffer pools in the deserializer code to
allow buffer reuse.

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

@@ -75,22 +75,6 @@ public abstract class AbstractByteArraySerializer implements
 		}
 	}
 
-	/**
-	 * Copy size bytes to a new buffer exactly size bytes long.
-	 * @param buffer The buffer containing the data.
-	 * @param size The number of bytes to copy.
-	 * @return The new buffer, or the buffer parameter if it is
-	 * already the correct size.
-	 */
-	protected byte[] copyToSizedArray(byte[] buffer, int size) {
-		if (size == buffer.length) {
-			return buffer;
-		}
-		byte[] assembledData = new byte[size];
-		System.arraycopy(buffer, 0, assembledData, 0, size);
-		return assembledData;
-	}
-
 	protected void publishEvent(Exception cause, byte[] buffer, int offset) {
 		TcpDeserializationExceptionEvent event = new TcpDeserializationExceptionEvent(this, cause, buffer, offset);
 		if (this.applicationEventPublisher != null) {

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

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2016 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 org.springframework.integration.util.SimplePool;
+import org.springframework.integration.util.SimplePool.PoolItemCallback;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for deserializers that cannot determine the buffer size needed.
+ * Optionally pools buffers.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public abstract class AbstractPooledBufferByteArraySerializer extends AbstractByteArraySerializer {
+
+	private SimplePool<byte[]> pool;
+
+	private long poolWaitTimeout = Long.MAX_VALUE;
+
+	/**
+	 * Set the pool size for deserialization buffers.
+	 * @param size the size, -1 for unlimited.
+	 * @since 4.3
+	 */
+	public void setPoolSize(int size) {
+		Assert.isNull(this.pool, "Cannot change pool size once set");
+		this.pool = new SimplePool<byte[]>(size, new PoolItemCallback<byte[]>() {
+
+			@Override
+			public byte[] createForPool() {
+				return new byte[getMaxMessageSize()];
+			}
+
+			@Override
+			public boolean isStale(byte[] item) {
+				return false; // never stale
+			}
+
+			@Override
+			public void removedFromPool(byte[] item) {
+			}
+
+		});
+		this.pool.setWaitTimeout(this.poolWaitTimeout);
+	}
+
+	/**
+	 * Set the pool wait timeout if a pool is configured, default unlimited.
+	 * @param poolWaitTimeout the timeout.
+	 */
+	public void setPoolWaitTimeout(long poolWaitTimeout) {
+		this.poolWaitTimeout = poolWaitTimeout;
+		if (this.pool != null) {
+			this.pool.setWaitTimeout(poolWaitTimeout);
+		}
+	}
+
+	@Override
+	public final byte[] deserialize(InputStream inputStream) throws IOException {
+		byte[] buffer = this.pool == null ? new byte[this.maxMessageSize] : this.pool.getItem();
+		try {
+			return doDeserialize(inputStream, buffer);
+		}
+		finally {
+			if (this.pool != null) {
+				this.pool.releaseItem(buffer);
+			}
+		}
+	}
+
+	/**
+	 * @param inputStream the input stream.
+	 * @param buffer the raw working buffer (maxMessageSize).
+	 * @return the decoded bytes.
+	 * @throws IOException an io exception.
+	 * @since 4.3
+	 */
+	protected abstract byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException;
+
+	/**
+	 * Copy size bytes to a new buffer exactly size bytes long. If a pool is not
+	 * in use and the array is already the correct length, it is simply returned.
+	 * @param buffer The buffer containing the data.
+	 * @param size The number of bytes to copy.
+	 * @return The new buffer, or the buffer parameter if it is
+	 * already the correct size and there is no pool.
+	 */
+	protected byte[] copyToSizedArray(byte[] buffer, int size) {
+		if (size == buffer.length && this.pool == null) {
+			return buffer;
+		}
+		byte[] assembledData = new byte[size];
+		System.arraycopy(buffer, 0, assembledData, 0, size);
+		return assembledData;
+	}
+
+}

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

@@ -28,7 +28,7 @@ import java.io.OutputStream;
  * @author Gary Russell
  * @since 2.0
  */
-public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer {
+public class ByteArrayCrLfSerializer extends AbstractPooledBufferByteArraySerializer {
 
 	private static final byte[] CRLF = "\r\n".getBytes();
 
@@ -39,8 +39,7 @@ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer {
 	 * being read).
 	 */
 	@Override
-	public byte[] deserialize(InputStream inputStream) throws IOException {
-		byte[] buffer = new byte[this.maxMessageSize];
+	public byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
 		int n = this.fillToCrLf(inputStream, buffer);
 		return this.copyToSizedArray(buffer, n);
 	}

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

@@ -41,7 +41,7 @@ import java.net.SocketTimeoutException;
  * @since 2.0.3
  *
  */
-public class ByteArrayRawSerializer extends AbstractByteArraySerializer {
+public class ByteArrayRawSerializer extends AbstractPooledBufferByteArraySerializer {
 
 	private final boolean treatTimeoutAsEndOfMessage;
 
@@ -66,8 +66,7 @@ public class ByteArrayRawSerializer extends AbstractByteArraySerializer {
 	}
 
 	@Override
-	public byte[] deserialize(InputStream inputStream) throws IOException {
-		byte[] buffer = new byte[this.maxMessageSize];
+	protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
 		int n = 0;
 		int bite = 0;
 		if (logger.isDebugEnabled()) {
@@ -90,15 +89,13 @@ public class ByteArrayRawSerializer extends AbstractByteArraySerializer {
 					}
 					break;
 				}
-				buffer[n++] = (byte) bite;
 				if (n >= this.maxMessageSize) {
 					throw new IOException("Socket was not closed before max message length: "
 							+ this.maxMessageSize);
 				}
+				buffer[n++] = (byte) bite;
 			}
-			byte[] assembledData = new byte[n];
-			System.arraycopy(buffer, 0, assembledData, 0, n);
-			return assembledData;
+			return copyToSizedArray(buffer, n);
 		}
 		catch (SoftEndOfStreamException e) {
 			throw e;

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

@@ -28,7 +28,7 @@ import java.io.OutputStream;
  * @author Gary Russell
  * @since 2.2
  */
-public class ByteArraySingleTerminatorSerializer extends AbstractByteArraySerializer {
+public class ByteArraySingleTerminatorSerializer extends AbstractPooledBufferByteArraySerializer {
 
 	private final byte terminator;
 
@@ -43,8 +43,7 @@ public class ByteArraySingleTerminatorSerializer extends AbstractByteArraySerial
 	 * being read).
 	 */
 	@Override
-	public byte[] deserialize(InputStream inputStream) throws IOException {
-		byte[] buffer = new byte[this.maxMessageSize];
+	protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
 		int n = 0;
 		int bite;
 		if (logger.isDebugEnabled()) {
@@ -67,9 +66,7 @@ public class ByteArraySingleTerminatorSerializer extends AbstractByteArraySerial
 							+ this.maxMessageSize);
 				}
 			}
-			byte[] assembledData = new byte[n];
-			System.arraycopy(buffer, 0, assembledData, 0, n);
-			return assembledData;
+			return copyToSizedArray(buffer, n);
 		}
 		catch (SoftEndOfStreamException e) {
 			throw e;

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

@@ -30,7 +30,7 @@ import org.springframework.integration.mapping.MessageMappingException;
  * @author Gary Russell
  * @since 2.0
  */
-public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer {
+public class ByteArrayStxEtxSerializer extends AbstractPooledBufferByteArraySerializer {
 
 	public static final int STX = 0x02;
 
@@ -45,18 +45,16 @@ public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer {
 	 *
 	 */
 	@Override
-	public byte[] deserialize(InputStream inputStream) throws IOException {
+	public byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
 		int bite = inputStream.read();
 		if (bite < 0) {
 			throw new SoftEndOfStreamException("Stream closed between payloads");
 		}
-		byte[] buffer = null;
 		int n = 0;
 		try {
 			if (bite != STX) {
 				throw new MessageMappingException("Expected STX to begin message");
 			}
-			buffer = new byte[this.maxMessageSize];
 			while ((bite = inputStream.read()) != ETX) {
 				checkClosure(bite);
 				buffer[n++] = (byte) bite;
@@ -65,9 +63,7 @@ public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer {
 							+ this.maxMessageSize);
 				}
 			}
-			byte[] assembledData = new byte[n];
-			System.arraycopy(buffer, 0, assembledData, 0, n);
-			return assembledData;
+			return copyToSizedArray(buffer, n);
 		}
 		catch (IOException e) {
 			publishEvent(e, buffer, n);

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2016 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.assertNotSame;
+import static org.junit.Assert.fail;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.util.Set;
+
+import org.junit.Test;
+
+import org.springframework.integration.test.util.TestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public class PooledDeserializationTests {
+
+	@Test
+	public void testCRLF() throws IOException {
+		ByteArrayCrLfSerializer deser = new ByteArrayCrLfSerializer();
+		deser.setPoolSize(2);
+		ByteArrayInputStream bais = new ByteArrayInputStream("foo\r\n".getBytes());
+		for (int i = 0; i < 5; i++) {
+			bais.reset();
+			byte[] bytes = deser.deserialize(bais);
+			assertEquals("foo", new String(bytes));
+		}
+		try {
+			deser.deserialize(bais);
+			fail("Expected SoftEndOfStreamException");
+		}
+		catch (SoftEndOfStreamException e) {
+			// expected
+		}
+		assertEquals(1, TestUtils.getPropertyValue(deser, "pool.allocated", Set.class).size());
+		assertEquals(0, TestUtils.getPropertyValue(deser, "pool.inUse", Set.class).size());
+	}
+
+	@Test
+	public void testRawMaxMessageSizeEqualDontReturnPooledItem() throws IOException {
+		ByteArrayRawSerializer deser = new ByteArrayRawSerializer();
+		deser.setPoolSize(2);
+		deser.setMaxMessageSize(3);
+		ByteArrayInputStream bais = new ByteArrayInputStream("foo".getBytes());
+		byte[] bytes = deser.deserialize(bais);
+		assertEquals("foo", new String(bytes));
+		assertEquals(1, TestUtils.getPropertyValue(deser, "pool.allocated", Set.class).size());
+		assertEquals(0, TestUtils.getPropertyValue(deser, "pool.inUse", Set.class).size());
+		assertNotSame(bytes, TestUtils.getPropertyValue(deser, "pool.allocated", Set.class).iterator().next());
+	}
+
+}

src/reference/asciidoc/ip.adoc

@@ -243,14 +243,14 @@ Connection factories are configured to use (de)serializers to convert between th
 This is accomplished by providing a deserializer and serializer for inbound and outbound messages respectively.
 A number of standard (de)serializers are provided.
 
-The `ByteArrayCrlfSerializer`, converts a byte array to a stream of bytes followed by carriage return and linefeed characters (\r\n).
+The `ByteArrayCrlfSerializer`^*^, converts a byte array to a stream of bytes followed by carriage return and linefeed characters (\r\n).
 This is the default (de)serializer and can be used with telnet as a client, for example.
 
-The `ByteArraySingleTerminatorSerializer`, converts a byte array to a stream of bytes followed by a single termination character (default 0x00).
+The `ByteArraySingleTerminatorSerializer`^*^, converts a byte array to a stream of bytes followed by a single termination character (default 0x00).
 
-The `ByteArrayLfSerializer`, converts a byte array to a stream of bytes followed by a single linefeed character (0x0a).
+The `ByteArrayLfSerializer`^*^, converts a byte array to a stream of bytes followed by a single linefeed character (0x0a).
 
-The `ByteArrayStxEtxSerializer`, converts a byte array to a stream of bytes preceded by an STX (0x02) and followed by an ETX (0x03).
+The `ByteArrayStxEtxSerializer`^*^, converts a byte array to a stream of bytes preceded by an STX (0x02) and followed by an ETX (0x03).
 
 The `ByteArrayLengthHeaderSerializer`, 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.
@@ -260,7 +260,7 @@ However, the length header can be a single byte (unsigned) for messages up to 25
 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`, converts a byte array to a stream of bytes and adds no additional message demarcation data; with this (de)serializer, the end of a message is indicated by the client closing the socket in an orderly fashion.
+The `ByteArrayRawSerializer`^*^, converts a byte array to a stream of bytes and adds no additional message demarcation data; with this (de)serializer, the end of a message is indicated by the client closing the socket in an orderly fashion.
 When using this serializer, message reception will hang until the client closes the socket, or a timeout occurs; a timeout will NOT result in a message.
 When this serializer is being used, and the client is a Spring Integration application, the client must use a connection factory that is configured with single-use=true - this causes the adapter to close the socket after sending the message; the serializer will not, itself, close the connection.
 This serializer should only be used with connection factories used by channel adapters (not gateways), and the connection factories should be used by either an inbound or outbound adapter, and not both.
@@ -280,6 +280,20 @@ If the size is exceeded by an incoming message, an exception will be thrown.
 The default maximum message size is 2048 bytes, and can be increased by setting the `maxMessageSize` property.
 If you are using the default (de)serializer and wish to increase the maximum message size, you must declare it as an explicit bean with the property set and configure the connection factory to use that bean.
 
+The classes marked with ^*^ above use an intermediate buffer and copy the decoded data to a final buffer of the correct
+size.
+Starting with _version 4.3_, these can be configured with a `poolSize` property to allow these raw buffers to be reused
+instead of being allocated and discarded for each message, which is the default behavior.
+Setting the property to a negative value will create a pool that has no bounds.
+If the pool is bounded, you can also set the `poolWaitTimeout` property (milliseconds) after which an exception is
+thrown if no buffer becomes available; it defaults to infinity.
+Such an exception will cause the socket to be closed.
+
+If you wish to use the same mechanism in custom deserializers, subclass `AbstractPooledBufferByteArraySerializer`
+instead of its super class `AbstractByteArraySerializer`, and implement `doDeserialize()` instead of `deserialize()`.
+The buffer will be returned to the pool automatically.
+`AbstractPooledBufferByteArraySerializer` also provides a convenient utility method `copyToSizedArray()`.
+
 The `MapJsonSerializer` uses a Jackson `ObjectMapper` to convert between a `Map` and JSON.
 This can be used in conjunction with a `MessageConvertingTcpMessageMapper` and a `MapMessageConverter` to transfer selected headers and the payload in a JSON format.
 

src/reference/asciidoc/whats-new.adoc

@@ -100,12 +100,21 @@ for more information.
 
 ==== TCP/UDP Changes
 
+===== Events
+
 A new `TcpConnectionServerListeningEvent` is emitted when a server connection factory is started.
 See <<tcp-events>> for more information.
 
 The `destination-expression` and `socket-expression` are now available for the `<int-ip:udp-outbound-channel-adapter>`.
 See <<udp-adapters>> for more information.
 
+===== Stream Deserializers
+
+The various deserializers that can't allocate the final buffer until the whole message has been assembled now support
+pooling of the raw buffer into which the data is received, rather than creating and discarding a buffer for each
+message.
+See <<connection-factories>> for more information.
+
 ==== File Changes
 
 ===== Destination Directory Creation