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 27c677380f..3bf72874fd 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 @@ -125,9 +125,16 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling @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 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling * @since 5.2 * @see Socket#shutdownInput() */ + @Override public void shutdownInput() throws IOException { this.socket.shutdownInput(); } @@ -301,6 +309,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling * @since 5.2 * @see Socket#shutdownOutput() */ + @Override public void shutdownOutput() throws IOException { this.socket.shutdownOutput(); } 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 23cc944062..868d8b0a6e 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 @@ -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 @@ import java.io.IOException; * @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); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index db2ff0bdb0..d8ccc01ae0 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -282,8 +282,7 @@ public class TcpOutboundGatewayTests { fail("Unexpected " + e.getMessage()); } else { - assertThat(e.getCause()).isNotNull(); - assertThat(e.getCause() instanceof MessageTimeoutException).isTrue(); + assertThat(e.getCause()).isInstanceOf(MessageTimeoutException.class); } timeouts++; continue; 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 ac540b0a08..1c2d95bb5d 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 @@ -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 class TcpNetConnectionTests { 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(); + } + } diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 714cdab6fb..b8019c5143 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index f869153bcd..a324fadd67 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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