diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index d80e080fb5..43c1810f58 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/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 buffers = new LinkedBlockingQueue(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(); + } + + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index 0b60f6a1a5..930ac2ef56 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java index 4a39440e8d..4b0c7e76c7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java +++ b/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(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index 8cd44f0ff6..f91c0eb134 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/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() { 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; } });