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

Fixes #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

Author: garyrussell

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

@@ -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 void close() {
 		try {
 			this.socket.close();
 		}
-		catch (Exception e) { }
+		catch (Exception e) {
+		}
 		super.close();
 	}
 
@@ -117,7 +120,15 @@ public synchronized void send(Message<?> message) throws Exception {
 
 	@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 void run() {
 				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 @@ protected boolean handleReadException(Exception e) {
 					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);

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,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.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 boolean onMessage(Message<?> message) {
 		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();
+	}
+
 }