diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index a04dfce7be..2a49b7d21b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -120,7 +120,15 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling @Override public Object getPayload() throws Exception { - return this.getDeserializer().deserialize(inputStream()); + InputStream inputStream; + try { + inputStream = inputStream(); + } + catch (IOException e1) { + throw new SoftEndOfStreamException("Socket closed when getting input stream", e1); + } + return getDeserializer() + .deserialize(inputStream); } @Override diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java index a76a3ccf6a..13138de8b5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java +++ b/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); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java index c1494b8231..6fadccbd38 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java @@ -16,6 +16,8 @@ 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; @@ -23,22 +25,30 @@ 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 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 = 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(); + } + }