INT-2876 Use Custom InputStream For NIO

Previously TcpNioConnection used a pair of Piped(In|Out)putStreams
to pass data from the read event to the deserializer.

However, it is possible to get a 'Broken Pipe' exception if the
last thread that wrote to the pipe terminates.

Replace the use of this pair of streams with an inner class
ChannelInputStream, which provides the same functionality, but
with no dependency on threading.

Uses a BlockingQueue to store read buffers which are consumed by
the message assembler thread (read from the InputStream in the
Deserializer). To avoid OOM conditions, the queue is limited to
5 "unread" buffers; and a timeout will occur if the reader
doesn't consume some data to free up space in the queue.

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java

@@ -17,18 +17,19 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.OutputStream;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
 import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -53,9 +54,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private final ChannelOutputStream channelOutputStream;
 
-	private volatile PipedOutputStream pipedOutputStream;
-
-	private volatile PipedInputStream pipedInputStream;
+	private final ChannelInputStream channelInputStream = new ChannelInputStream();
 
 	private volatile boolean usingDirectBuffers;
 
@@ -88,8 +87,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		if (receiveBufferSize <= 0) {
 			receiveBufferSize = this.maxMessageSize;
 		}
-		this.pipedInputStream = new PipedInputStream(receiveBufferSize);
-		this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream);
 		this.channelOutputStream = new ChannelOutputStream();
 	}
 
@@ -103,11 +100,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	private void doClose() {
-		if (pipedOutputStream != null) {
-			try {
-				pipedOutputStream.close();
-			} catch (IOException e) {}
-		}
+		try {
+			channelInputStream.close();
+		} catch (IOException e) {}
 		try {
 			this.socketChannel.close();
 		} catch (Exception e) {}
@@ -129,7 +124,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	public Object getPayload() throws Exception {
-		return this.getDeserializer().deserialize(pipedInputStream);
+		return this.getDeserializer().deserialize(this.channelInputStream);
 	}
 
 	public int getPort() {
@@ -197,7 +192,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 				return;
 			}
 		} finally {
-			logger.trace("Nio message assembler exiting...");
+			if (logger.isTraceEnabled()) {
+				logger.trace(this.getConnectionId() + " Nio message assembler exiting...");
+			}
 			// Final check in case new data came in and the
 			// timing was such that we were the last assembler and
 			// a new one wasn't run
@@ -212,7 +209,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	private boolean dataAvailable() throws IOException {
-		return this.pipedInputStream.available() > 0 || writingToPipe;
+		return this.channelInputStream.available() > 0 || writingToPipe;
 	}
 
 	/**
@@ -325,7 +322,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
 				}
 			});
 			if (!latch.await(this.pipeTimeout , TimeUnit.MILLISECONDS)) {
-				throw new MessagingException("Timed out writing to pipe, probably due to insufficient threads in " +
+				this.close();
+				throw new MessagingException("Timed out writing to ChannelInputStream, probably due to insufficient threads in " +
 						"a fixed thread pool; consider increasing this task executor pool size");
 			}
 		} finally {
@@ -336,10 +334,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	protected void sendToPipe(ByteBuffer rawBuffer) throws IOException {
 		Assert.notNull(rawBuffer, "rawBuffer cannot be null");
 		if (logger.isTraceEnabled()) {
-			logger.trace("Sending " + rawBuffer.limit() + " to pipe");
+			logger.trace(this.getConnectionId() + " Sending " + rawBuffer.limit() + " to pipe");
 		}
-		this.pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
-		this.pipedOutputStream.flush();
+		this.channelInputStream.write(rawBuffer.array(), rawBuffer.limit());
 		rawBuffer.clear();
 	}
 
@@ -498,4 +495,98 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	}
 
+	/**
+	 * Provides an InputStream to receive data from {@link SocketChannel#read(ByteBuffer)}
+	 * operations. Each new buffer is added to a BlockingQueue; when the reading thread
+	 * exhausts the current buffer, it retrieves the next from the queue.
+	 * Writes block for up to the pipeTimeout if 5 buffers are queued to be read.
+	 *
+	 */
+	class ChannelInputStream extends InputStream {
+
+		private static final int BUFFER_LIMIT = 5;
+
+		private final BlockingQueue<byte[]> buffers = new LinkedBlockingQueue<byte[]>(BUFFER_LIMIT);
+
+		private volatile byte[] currentBuffer;
+
+		private volatile int currentOffset;
+
+		private final AtomicInteger available = new AtomicInteger();
+
+		private volatile boolean isClosed;
+
+		@Override
+		public synchronized int read() throws IOException {
+			if (this.isClosed && available.get() == 0) {
+				return -1;
+			}
+			if (this.currentBuffer == null) {
+				this.currentBuffer = getNextBuffer();
+				this.currentOffset = 0;
+				if (this.currentBuffer == null) {
+					return -1;
+				}
+			}
+			int bite;
+			bite = this.currentBuffer[this.currentOffset++];
+			this.available.decrementAndGet();
+			if (this.currentOffset >= this.currentBuffer.length) {
+				this.currentBuffer = null;
+			}
+			return bite;
+		}
+
+		private byte[] getNextBuffer() throws IOException {
+			byte[] buffer = null;
+			while (buffer == null) {
+				try {
+					buffer = buffers.poll(1, TimeUnit.SECONDS);
+					if (buffer == null && this.isClosed) {
+						return null;
+					}
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					throw new IOException("Interrupted while waiting for data", e);
+				}
+			}
+			return buffer;
+		}
+
+		/**
+		 * Blocks if the blocking queue already contains 5 buffers.
+		 * @param array
+		 * @param bytesToWrite
+		 * @throws IOException
+		 */
+		public void write(byte[] array, int bytesToWrite) throws IOException {
+			if (bytesToWrite > 0) {
+				byte[] buffer = new byte[bytesToWrite];
+				System.arraycopy(array, 0, buffer, 0, bytesToWrite);
+				this.available.addAndGet(bytesToWrite);
+				try {
+					if (!this.buffers.offer(buffer, pipeTimeout, TimeUnit.MILLISECONDS)) {
+						throw new IOException("Timed out waiting for buffer space");
+					}
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					throw new IOException("Interrupted while waiting for buffer space", e);
+				}
+			}
+		}
+
+		@Override
+		public void close() throws IOException {
+			super.close();
+			this.isClosed = true;
+		}
+
+		@Override
+		public int available() throws IOException {
+			return this.available.get();
+		}
+
+	}
 }

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

@@ -86,6 +86,7 @@ public class ConnectionToConnectionTests {
 			TcpConnection connection = client.getConnection();
 			connection.send(MessageBuilder.withPayload("Test").build());
 			Message<?> message = serverSideChannel.receive(10000);
+			assertNotNull(message);
 			MessageHistory history = MessageHistory.read(message);
 			//org.springframework.integration.test.util.TestUtils
 			Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
@@ -105,6 +106,7 @@ public class ConnectionToConnectionTests {
 		TcpConnection connection = client.getConnection();
 		connection.send(MessageBuilder.withPayload("Test").build());
 		Message<?> message = serverSideChannel.receive(10000);
+		assertNotNull(message);
 		MessageHistory history = MessageHistory.read(message);
 		//org.springframework.integration.test.util.TestUtils
 		Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);

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

@@ -45,84 +45,84 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
 public class TcpConfigInboundGatewayTests {
-	
+
 	static AbstractApplicationContext staticContext;
-	
+
 	@Autowired
 	AbstractApplicationContext ctx;
-	
+
 	@Autowired
 	@Qualifier(value="crLfServer")
 	AbstractServerConnectionFactory crLfServer;
-	
+
 	@Autowired
 	@Qualifier(value="stxEtxServer")
 	AbstractServerConnectionFactory stxEtxServer;
-	
+
 	@Autowired
 	@Qualifier(value="lengthHeaderServer")
 	AbstractServerConnectionFactory lengthHeaderServer;
-	
+
 	@Autowired
 	@Qualifier(value="javaSerialServer")
 	AbstractServerConnectionFactory javaSerialServer;
-	
+
 	@Autowired
 	@Qualifier(value="crLfClient")
 	AbstractClientConnectionFactory crLfClient;
-	
+
 	@Autowired
 	@Qualifier(value="stxEtxClient")
 	AbstractClientConnectionFactory stxEtxClient;
-	
+
 	@Autowired
 	@Qualifier(value="lengthHeaderClient")
 	AbstractClientConnectionFactory lengthHeaderClient;
-	
+
 	@Autowired
 	@Qualifier(value="javaSerialClient")
 	AbstractClientConnectionFactory javaSerialClient;
-	
+
 	@Autowired
 	@Qualifier(value="crLfServerNio")
 	AbstractServerConnectionFactory crLfServerNio;
-	
+
 	@Autowired
 	@Qualifier(value="stxEtxServerNio")
 	AbstractServerConnectionFactory stxEtxServerNio;
-	
+
 	@Autowired
 	@Qualifier(value="lengthHeaderServerNio")
 	AbstractServerConnectionFactory lengthHeaderServerNio;
-	
+
 	@Autowired
 	@Qualifier(value="javaSerialServerNio")
 	AbstractServerConnectionFactory javaSerialServerNio;
-	
+
 	@Autowired
 	@Qualifier(value="crLfClientNio")
 	AbstractClientConnectionFactory crLfClientNio;
-	
+
 	@Autowired
 	@Qualifier(value="stxEtxClientNio")
 	AbstractClientConnectionFactory stxEtxClientNio;
-	
+
 	@Autowired
 	@Qualifier(value="lengthHeaderClientNio")
 	AbstractClientConnectionFactory lengthHeaderClientNio;
-	
+
 	@Autowired
 	@Qualifier(value="javaSerialClientNio")
 	AbstractClientConnectionFactory javaSerialClientNio;
-	
+
 	@Autowired
 	@Qualifier(value="gatewayCrLf")
 	TcpInboundGateway gatewayCrLf;
-	
+
 	@Autowired
 	@Qualifier(value="gatewayStxEtx")
 	TcpInboundGateway gatewayStxEtx;
-	
+
 	@Autowired
 	@Qualifier(value="gatewayLength")
 	TcpInboundGateway gatewayLength;
@@ -134,11 +134,11 @@ public class TcpConfigInboundGatewayTests {
 	@Autowired
 	@Qualifier(value="gatewayCrLfNio")
 	TcpInboundGateway gatewayCrLfNio;
-	
+
 	@Autowired
 	@Qualifier(value="gatewayStxEtxNio")
 	TcpInboundGateway gatewayStxEtxNio;
-	
+
 	@Autowired
 	@Qualifier(value="gatewayLengthNio")
 	TcpInboundGateway gatewayLengthNio;
@@ -156,8 +156,8 @@ public class TcpConfigInboundGatewayTests {
 
 	@Test
 	public void testCrLfNio() throws Exception {
-		waitListening(gatewayCrLf);
-		Socket socket = SocketFactory.getDefault().createSocket("localhost", crLfServer.getPort());
+		waitListening(gatewayCrLfNio);
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", crLfServerNio.getPort());
 		crLfGuts(socket);
 	}
 
@@ -285,7 +285,7 @@ public class TcpConfigInboundGatewayTests {
 				throw new Exception("Gateway failed to listen");
 			}
 		}
-		
+
 	}
 
 	@Before
@@ -294,7 +294,7 @@ public class TcpConfigInboundGatewayTests {
 			staticContext = ctx;
 		}
 	}
-	
+
 	@AfterClass
 	public static void shutDown() {
 		staticContext.close();

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java

@@ -26,7 +26,6 @@ import static org.mockito.Mockito.when;
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.net.ServerSocket;
 import java.net.Socket;
@@ -62,8 +61,6 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.ReflectionUtils.FieldCallback;
 import org.springframework.util.ReflectionUtils.FieldFilter;
-import org.springframework.util.ReflectionUtils.MethodCallback;
-import org.springframework.util.ReflectionUtils.MethodFilter;
 
 
 /**
@@ -260,29 +257,26 @@ public class TcpNioConnectionTests {
 				doAnswer(new Answer<Integer>() {
 					public Integer answer(InvocationOnMock invocation) throws Throwable {
 						ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0];
-						buffer.position(1025);
-						return 1025;
+						buffer.position(1);
+						return 1;
 					}
 				}).when(channel).read(Mockito.any(ByteBuffer.class));
 				when(socket.getReceiveBufferSize()).thenReturn(1024);
 				final TcpNioConnection connection = new TcpNioConnection(channel, false, false);
 				connection.setTaskExecutor(exec);
 				connection.setPipeTimeout(200);
-				ReflectionUtils.doWithMethods(TcpNioConnection.class, new MethodCallback() {
-					public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
-						method.setAccessible(true);
-						try {
-							method.invoke(connection, (Object[]) null);
-						}
-						catch (InvocationTargetException e) {
-							throw (RuntimeException) e.getCause();
-						}
+				Method method = TcpNioConnection.class.getDeclaredMethod("doRead");
+				method.setAccessible(true);
+				// Nobody reading, should timeout on 6th write.
+				try {
+					for (int i = 0; i < 6; i++) {
+						method.invoke(connection);
 					}
-				}, new MethodFilter() {
-					public boolean matches(Method method) {
-						return method.getName().equals("doRead");
-					}
-				});
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+					throw (Exception) e.getCause();
+				}
 				return null;
 			}
 		});
@@ -291,7 +285,7 @@ public class TcpNioConnectionTests {
 			fail("Expected exception, got " + o);
 		}
 		catch (ExecutionException e) {
-			assertEquals("Timed out writing to pipe, probably due to insufficient threads in " +
+			assertEquals("Timed out writing to ChannelInputStream, probably due to insufficient threads in " +
 					"a fixed thread pool; consider increasing this task executor pool size", e.getCause()
 					.getMessage());
 		}
@@ -319,28 +313,23 @@ public class TcpNioConnectionTests {
 				connection.setTaskExecutor(exec);
 				connection.registerListener(new TcpListener(){
 					public boolean onMessage(Message<?> message) {
-						System.out.println(message);
 						messageLatch.countDown();
 						return false;
 					}
 				});
 				connection.setMapper(new TcpMessageMapper());
 				connection.setDeserializer(new ByteArrayCrLfSerializer());
-				ReflectionUtils.doWithMethods(TcpNioConnection.class, new MethodCallback() {
-					public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
-						method.setAccessible(true);
-						try {
-							method.invoke(connection, (Object[]) null);
-						}
-						catch (InvocationTargetException e) {
-							throw (RuntimeException) e.getCause();
-						}
+				Method method = TcpNioConnection.class.getDeclaredMethod("doRead");
+				method.setAccessible(true);
+				try {
+					for (int i = 0; i < 20; i++) {
+						method.invoke(connection);
 					}
-				}, new MethodFilter() {
-					public boolean matches(Method method) {
-						return method.getName().equals("doRead");
-					}
-				});
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+					throw (Exception) e.getCause();
+				}
 				return null;
 			}
 		});