GH-2974: Fix race in TcpNetConnection.getPayload()

Fixes https://github.com/spring-projects/spring-integration/issues/2974

There is a race in that we could get a `SocketException` in `inputStream`.
Since this is between payloads, it needs to be thrown as a
`SoftEndOfStreamException`.

Also fix unnecessary `this.` in `MessageHistoryConfigurer.java`.

**cherry-pick to 5.0.x, 4.3.x**

* * Add javadocs to SoftEndOfStreamException

# Conflicts:
#	spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java

# Conflicts:
#	spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
#	spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java
This commit is contained in:
Gary Russell
2019-06-26 11:17:44 -04:00
committed by Artem Bilan
parent 3c69b2664b
commit 2e7c5d82f2
3 changed files with 89 additions and 13 deletions

View File

@@ -17,6 +17,8 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
@@ -82,7 +84,8 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
try {
this.socket.close();
}
catch (Exception e) { }
catch (Exception e) {
}
super.close();
}
@@ -117,7 +120,15 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
@Override
public Object getPayload() throws Exception {
return this.getDeserializer().deserialize(this.socket.getInputStream());
InputStream inputStream;
try {
inputStream = this.socket.getInputStream();
}
catch (IOException e1) {
throw new SoftEndOfStreamException("Socket closed when getting input stream", e1);
}
return getDeserializer()
.deserialize(inputStream);
}
@Override
@@ -184,9 +195,9 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
catch (NoListenerException nle) { // could also be thrown by an interceptor
if (logger.isWarnEnabled()) {
logger.warn("Unexpected message - no endpoint registered with connection interceptor: "
+ getConnectionId()
+ " - "
+ message);
+ getConnectionId()
+ " - "
+ message);
}
}
catch (Exception e2) {
@@ -230,24 +241,24 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
if (noReadErrorOnClose) {
if (logger.isTraceEnabled()) {
logger.trace("Read exception " +
this.getConnectionId(), e);
this.getConnectionId(), e);
}
else if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
}
else if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
this.getConnectionId(), e);
}
else {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
}
this.sendExceptionToListener(e);

View File

@@ -30,12 +30,29 @@ public class SoftEndOfStreamException extends IOException {
private static final long serialVersionUID = 7309907445617226978L;
/**
* Default constructor.
*/
public SoftEndOfStreamException() {
super();
}
/**
* Construct an instance with the message.
* @param message the message.
*/
public SoftEndOfStreamException(String message) {
super(message);
}
/**
* Construct an instance with the message and cause.
* @param message the message.
* @param cause the cause.
* @since 4.3.21.
*/
public SoftEndOfStreamException(String message, Throwable cause) {
super(message, cause);
}
}

View File

@@ -16,21 +16,30 @@
package org.springframework.integration.ip.tcp.connection;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.SocketFactory;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.mockito.Mockito;
@@ -43,6 +52,7 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.converter.MapMessageConverter;
import org.springframework.integration.test.util.TestUtils;
@@ -157,4 +167,42 @@ public class TcpNetConnectionTests {
assertEquals("baz", inboundMessage.get().getHeaders().get("bar"));
}
@Test
public void socketClosedNextRead() throws InterruptedException, IOException {
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0);
AtomicInteger port = new AtomicInteger();
CountDownLatch latch = new CountDownLatch(1);
ApplicationEventPublisher publisher = new ApplicationEventPublisher() {
@Override
public void publishEvent(Object ev) {
if (ev instanceof TcpConnectionServerListeningEvent) {
port.set(((TcpConnectionServerListeningEvent) ev).getPort());
latch.countDown();
}
}
@Override
public void publishEvent(ApplicationEvent event) {
publishEvent((Object) event);
}
};
server.setApplicationEventPublisher(publisher);
server.registerListener(message -> false);
server.afterPropertiesSet();
server.start();
assertTrue(latch.await(10, TimeUnit.SECONDS));
Socket socket = SocketFactory.getDefault().createSocket("localhost", port.get());
TcpNetConnection connection = new TcpNetConnection(socket, false, false, publisher, "socketClosedNextRead");
socket.close();
try {
connection.getPayload();
}
catch (Exception e) {
assertThat(e, instanceOf(SoftEndOfStreamException.class));
}
server.stop();
}
}