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

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

@@ -125,9 +125,16 @@ public synchronized void send(Message<?> message) {
 
 	@Override
 	public Object getPayload() {
+		InputStream inputStream;
+		try {
+			inputStream = inputStream();
+		}
+		catch (IOException e1) {
+			throw new SoftEndOfStreamException("Socket closed when getting input stream", e1);
+		}
 		try {
 			return getDeserializer()
-					.deserialize(inputStream());
+					.deserialize(inputStream);
 		}
 		catch (IOException e) {
 			throw new UncheckedIOException(e);
@@ -291,6 +298,7 @@ else if (readErrorOnClose) {
 	 * @since 5.2
 	 * @see Socket#shutdownInput()
 	 */
+	@Override
 	public void shutdownInput() throws IOException {
 		this.socket.shutdownInput();
 	}
@@ -301,6 +309,7 @@ public void shutdownInput() throws IOException {
 	 * @since 5.2
 	 * @see Socket#shutdownOutput()
 	 */
+	@Override
 	public void shutdownOutput() throws IOException {
 		this.socket.shutdownOutput();
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.ip.tcp.serializer;
 
-import java.io.IOException;
-
 /**
  * Used to communicate that a stream has closed, but between logical
  * messages.
@@ -26,16 +24,33 @@
  * @since 2.0
  *
  */
-public class SoftEndOfStreamException extends IOException {
+public class SoftEndOfStreamException extends RuntimeException {
 
-	private static final long serialVersionUID = 7309907445617226978L;
+	private static final long serialVersionUID = -2209857413498073058L;
 
+	/**
+	 * 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);
+	}
+
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java

@@ -282,8 +282,7 @@ public void testGoodNetTimeout() throws Exception {
 					fail("Unexpected " + e.getMessage());
 				}
 				else {
-					assertThat(e.getCause()).isNotNull();
-					assertThat(e.getCause() instanceof MessageTimeoutException).isTrue();
+					assertThat(e.getCause()).isInstanceOf(MessageTimeoutException.class);
 				}
 				timeouts++;
 				continue;

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java

@@ -17,28 +17,37 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.BDDMockito.given;
 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;
 
 import org.springframework.beans.DirectFieldAccessor;
+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;
@@ -137,4 +146,30 @@ public void transferHeaders() throws Exception {
 		assertThat(inboundMessage.get().getHeaders().get("bar")).isEqualTo("baz");
 	}
 
+	@Test
+	public void socketClosedNextRead() throws InterruptedException, IOException {
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0);
+		AtomicInteger port = new AtomicInteger();
+		CountDownLatch latch = new CountDownLatch(1);
+		ApplicationEventPublisher publisher = ev -> {
+			if (ev instanceof TcpConnectionServerListeningEvent) {
+				port.set(((TcpConnectionServerListeningEvent) ev).getPort());
+				latch.countDown();
+			}
+		};
+		server.setApplicationEventPublisher(publisher);
+		server.registerListener(message -> {
+			return false;
+		});
+		server.afterPropertiesSet();
+		server.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", port.get());
+		TcpNetConnection connection = new TcpNetConnection(socket, false, false, publisher, "socketClosedNextRead");
+		socket.close();
+		assertThatThrownBy(() -> connection.getPayload())
+			.isInstanceOf(SoftEndOfStreamException.class);
+		server.stop();
+	}
+
 }

src/reference/asciidoc/ip.adoc

@@ -498,6 +498,8 @@ To implement a custom serializer and deserializer pair, implement the `org.sprin
 When the deserializer detects a closed input stream between messages, it must throw a `SoftEndOfStreamException`; this is a signal to the framework to indicate that the close was "normal".
 If the stream is closed while decoding a message, some other exception should be thrown instead.
 
+Starting with version 5.2, `SoftEndOfStreamException` is now a `RuntimeException` instead of extending `IOException`.
+
 [[caching-cf]]
 ==== TCP Caching Client Connection Factory
 

src/reference/asciidoc/whats-new.adoc

@@ -87,6 +87,8 @@ See <<./ip.adoc#tcp-gateways, TCP Gateways>> for more information.
 The client connection factories now support `connectTimeout` which causes an exception to be thrown if the connection is not established in that time.
 See <<./ip.adoc#tcp-connection-factory, TCP Connection Factories>> for more information.
 
+`SoftEndOfStreamException` is now a `RuntimeException` instead of extending `IOException`.
+
 [[x5.2-mail]]
 ==== Mail Changes