From 93cd669b761fa48b264261c376132e0cf9bf1ea1 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 26 Jun 2019 11:17:44 -0400 Subject: [PATCH] GH-2974: Fix race in TcpNetConnection.getPayload() Fixes https://github.com/spring-projects/spring-integration/issues/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 --- .../ip/tcp/connection/TcpNetConnection.java | 10 +++++- .../serializer/SoftEndOfStreamException.java | 17 +++++++++ .../tcp/connection/TcpNetConnectionTests.java | 36 +++++++++++++++++++ 3 files changed, 62 insertions(+), 1 deletion(-) 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(); + } + }