GH-2974: Fix race in TcpNetConnection.getPayload() [51x]

spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java

@@ -154,8 +154,8 @@ private void trackComponentIfAny(TrackableComponent component) {
 		component.setShouldTrack(shouldTrack);
 		if (shouldTrack) {
 			this.currentlyTrackedComponents.add(component);
-			if (this.logger.isInfoEnabled()) {
-				this.logger.info("Enabling MessageHistory tracking for component '" + componentName + "'");
+			if (logger.isInfoEnabled()) {
+				logger.info("Enabling MessageHistory tracking for component '" + componentName + "'");
 			}
 		}
 	}
@@ -217,8 +217,8 @@ public void stop() {
 			if (this.running) {
 				this.currentlyTrackedComponents.forEach(component -> {
 					component.setShouldTrack(false);
-					if (this.logger.isInfoEnabled()) {
-						this.logger.info("Disabling MessageHistory tracking for component '"
+					if (logger.isInfoEnabled()) {
+						logger.info("Disabling MessageHistory tracking for component '"
 								+ component.getComponentName() + "'");
 					}
 				});

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

@@ -122,8 +122,15 @@ public synchronized void send(Message<?> message) throws Exception {
 
 	@Override
 	public Object getPayload() throws Exception {
+		InputStream inputStream;
+		try {
+			inputStream = inputStream();
+		}
+		catch (IOException e1) {
+			throw new SoftEndOfStreamException("Socket closed when getting input stream", e1);
+		}
 		return getDeserializer()
-				.deserialize(inputStream());
+				.deserialize(inputStream);
 	}
 
 	@Override

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

@@ -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);
+	}
+
 }

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

@@ -16,29 +16,39 @@
 
 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.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 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 +147,30 @@ public void transferHeaders() throws Exception {
 		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 = 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();
+	}
+
 }