From 57755a07ed3d47fc42c0ac4cc3410ef202169c21 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 21 Jun 2012 19:04:49 -0400 Subject: [PATCH] INT-2635 Set Buffer Size for Piped Streams Improve performance. For NIO sockets, a pair of PipedInput/OutputStreams are used to transfer data from the reading thread to the assembling thread. The stream used the default buffer size (1024) which was inefficient for large messages. This change uses the underlying socket's receiveBufferSize attribute to set the size of the piped stream, allowing for more efficient data transfer. Also adjusts the mocks in the deadlock detection test to ensure the received data is larger than the pipe buffer size. --- .../integration/ip/tcp/connection/TcpNioConnection.java | 6 +++++- .../ip/tcp/connection/TcpNioConnectionTests.java | 1 + 2 files changed, 6 insertions(+), 1 deletion(-) 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 5a5823b15d..b199707cdf 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 @@ -82,7 +82,11 @@ public class TcpNioConnection extends AbstractTcpConnection { public TcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost) throws Exception { super(socketChannel.socket(), server, lookupHost); this.socketChannel = socketChannel; - this.pipedInputStream = new PipedInputStream(); + int receiveBufferSize = socketChannel.socket().getReceiveBufferSize(); + if (receiveBufferSize <= 0) { + receiveBufferSize = this.maxMessageSize; + } + this.pipedInputStream = new PipedInputStream(receiveBufferSize); this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream); this.channelOutputStream = new ChannelOutputStream(); } 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 63ddeef49e..8cd44f0ff6 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 @@ -264,6 +264,7 @@ public class TcpNioConnectionTests { return 1025; } }).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);