From e4eac27b12752a964832da42acb1d7414564d669 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 28 Aug 2014 17:24:38 -0400 Subject: [PATCH] INT-3507 TCP - Use BufferedOutputStreams JIRA: https://jira.spring.io/browse/INT-3507 TcpNoDelay (false) helps to buffer IOs but only after the first write. Use `BufferedOutputStreams` for TCP writes. Use the socket `sendBufferSize` for the buffer size. --- .../ip/tcp/connection/TcpNetConnection.java | 19 +++++++++++++++++-- .../ip/tcp/connection/TcpNioConnection.java | 10 +++++++++- .../CachingClientConnectionFactoryTests.java | 11 ++++++++--- 3 files changed, 34 insertions(+), 6 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index dab0c96db7..79902e98fc 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -16,6 +16,8 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.BufferedOutputStream; +import java.io.OutputStream; import java.net.Socket; import java.net.SocketException; import java.net.SocketTimeoutException; @@ -23,8 +25,8 @@ import java.net.SocketTimeoutException; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; -import org.springframework.messaging.Message; import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException; +import org.springframework.messaging.Message; /** * A TcpConnection that uses and underlying {@link Socket}. @@ -37,6 +39,8 @@ public class TcpNetConnection extends TcpConnectionSupport { private final Socket socket; + private volatile OutputStream socketOutputStream; + private volatile long lastRead = System.currentTimeMillis(); private volatile long lastSend; @@ -72,16 +76,23 @@ public class TcpNetConnection extends TcpConnectionSupport { super.close(); } + @Override public boolean isOpen() { return !this.socket.isClosed(); } + @Override @SuppressWarnings("unchecked") public synchronized void send(Message message) throws Exception { + if (this.socketOutputStream == null) { + int writeBufferSize = this.socket.getSendBufferSize(); + this.socketOutputStream = new BufferedOutputStream(socket.getOutputStream(), + writeBufferSize > 0 ? writeBufferSize : 8192); + } Object object = this.getMapper().fromMessage(message); this.lastSend = System.currentTimeMillis(); try { - ((Serializer) this.getSerializer()).serialize(object, this.socket.getOutputStream()); + ((Serializer) this.getSerializer()).serialize(object, this.socketOutputStream); } catch (Exception e) { this.publishConnectionExceptionEvent(e); @@ -91,14 +102,17 @@ public class TcpNetConnection extends TcpConnectionSupport { this.afterSend(message); } + @Override public Object getPayload() throws Exception { return this.getDeserializer().deserialize(this.socket.getInputStream()); } + @Override public int getPort() { return this.socket.getPort(); } + @Override public Object getDeserializerStateKey() { try { return this.socket.getInputStream(); @@ -118,6 +132,7 @@ public class TcpNetConnection extends TcpConnectionSupport { * expires. If data is received on a single use socket with no listener, * a warning is logged. */ + @Override public void run() { boolean singleUse = this.isSingleUse(); TcpListener listener = this.getListener(); 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 8d236d1de1..f883208fed 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 @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.BufferedOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -60,6 +61,8 @@ public class TcpNioConnection extends TcpConnectionSupport { private final ChannelInputStream channelInputStream = new ChannelInputStream(); + private volatile OutputStream bufferedOutputStream; + private volatile boolean usingDirectBuffers; private volatile CompositeExecutor taskExecutor; @@ -133,10 +136,15 @@ public class TcpNioConnection extends TcpConnectionSupport { @SuppressWarnings("unchecked") public void send(Message message) throws Exception { synchronized(this.socketChannel) { + if (this.bufferedOutputStream == null) { + int writeBufferSize = this.socketChannel.socket().getSendBufferSize(); + this.bufferedOutputStream = new BufferedOutputStream(this.getChannelOutputStream(), + writeBufferSize > 0 ? writeBufferSize : 8192); + } Object object = this.getMapper().fromMessage(message); this.lastSend = System.currentTimeMillis(); try { - ((Serializer) this.getSerializer()).serialize(object, this.getChannelOutputStream()); + ((Serializer) this.getSerializer()).serialize(object, this.bufferedOutputStream); } catch (Exception e) { this.publishConnectionExceptionEvent(e); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index 820ed1a780..d14ea70ee7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -22,6 +22,8 @@ import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyInt; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -52,13 +54,13 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.TestingUtilities; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.DirtiesContext.ClassMode; import org.springframework.test.context.ContextConfiguration; @@ -208,6 +210,7 @@ public class CachingClientConnectionFactoryTests { assertEquals("Cached:" + mockConn2.toString(), conn2.toString()); cachingFactory.stop(); Answer answer = new Answer () { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; }}; @@ -345,7 +348,7 @@ public class CachingClientConnectionFactoryTests { Socket socket = mock(Socket.class); when(socket.isClosed()).thenReturn(true); // closed when next retrieved OutputStream stream = mock(OutputStream.class); - doThrow(new IOException("Foo")).when(stream).write(Mockito.any(byte[].class)); + doThrow(new IOException("Foo")).when(stream).write(any(byte[].class), anyInt(), anyInt()); when(socket.getOutputStream()).thenReturn(stream); TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() { @@ -409,6 +412,7 @@ public class CachingClientConnectionFactoryTests { final List connectionIds = new ArrayList(); final AtomicBoolean okToRun = new AtomicBoolean(true); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { while (okToRun.get()) { Message m = inbound.receive(1000); @@ -484,6 +488,7 @@ public class CachingClientConnectionFactoryTests { when(factory2.isActive()).thenReturn(true); doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; }