diff --git a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java index 575478a287..c965786c37 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java @@ -154,8 +154,8 @@ public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAwar 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 class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAwar 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() + "'"); } }); 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 8cdd3dc7ec..30aef5cf61 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 @@ -122,8 +122,15 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling @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 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..978542d5bc 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 5ebac71959..87805b4215 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(); + } + }