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 a47353a216..365e5e7d2f 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 @@ -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); 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 97af08ccd0..10ce90e50b 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,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(); + } + }