Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3507 TCP - Use BufferedOutputStreams

Browse Source 
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.
This commit is contained in:
Gary Russell
2014-08-28 17:24:38 -04:00
parent 0b917622a6
commit e4eac27b12
3 changed files with 34 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -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<Object>) this.getSerializer()).serialize(object, this.socket.getOutputStream());
((Serializer<Object>) 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();

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

@@ -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<Object>) this.getSerializer()).serialize(object, this.getChannelOutputStream());
((Serializer<Object>) this.getSerializer()).serialize(object, this.bufferedOutputStream);
}
catch (Exception e) {
this.publishConnectionExceptionEvent(e);

11
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -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<Object> answer = new Answer<Object> () {
@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<String> connectionIds = new ArrayList<String>();
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<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}