Merge pull request #702 from garyrussell/INT-2876

* INT-2876: Use Custom InputStream For NIO

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;
 			}
 		});