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
This commit is contained in:
Gary Russell
2019-06-26 11:17:44 -04:00
committed by Artem Bilan
parent 68dfbbe3f0
commit a75efd323a
4 changed files with 65 additions and 5 deletions

View File

@@ -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() + "'");
}
});

View File

@@ -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

View File

@@ -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);
}
}

View File

@@ -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();
}
}