From 466daa8774c8da78a971b48d096659a9c1f08524 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 5 Jun 2019 14:38:14 -0400 Subject: [PATCH] TCP: Connect Timeout; Close Stream - Add `connectTimeout` to client connection factories - Add `closeStreamAfterSend` to outbound gateway * Polishing - PR Comments. --- .../ip/config/IpAdapterParserUtils.java | 2 + .../TcpConnectionFactoryFactoryBean.java | 77 +++++++++++-------- .../ip/config/TcpConnectionFactoryParser.java | 2 + .../ip/config/TcpOutboundGatewayParser.java | 1 + .../ip/dsl/AbstractConnectionFactorySpec.java | 11 +++ .../dsl/TcpClientConnectionFactorySpec.java | 11 +++ .../ip/dsl/TcpOutboundGatewaySpec.java | 14 ++++ .../ip/tcp/TcpOutboundGateway.java | 22 +++++- .../AbstractClientConnectionFactory.java | 22 +++++- .../ip/tcp/connection/TcpConnection.java | 22 ++++++ .../TcpNetClientConnectionFactory.java | 5 +- .../ip/tcp/connection/TcpNetConnection.java | 20 +++++ .../TcpNioClientConnectionFactory.java | 17 +++- .../ip/tcp/connection/TcpNioConnection.java | 20 +++++ .../ip/config/spring-integration-ip-5.2.xsd | 17 ++++ ...ts-context.xml => ParserTests-context.xml} | 6 +- .../ip/config/ParserUnitTests.java | 32 ++++++-- .../ip/dsl/IpIntegrationTests.java | 45 +++++++++++ .../ip/tcp/TcpInboundGatewayTests.java | 58 ++++++++++++++ .../ip/tcp/connection/SocketSupportTests.java | 58 ++++++++++++-- src/reference/asciidoc/ip.adoc | 5 ++ src/reference/asciidoc/whats-new.adoc | 7 ++ 22 files changed, 420 insertions(+), 54 deletions(-) rename spring-integration-ip/src/test/java/org/springframework/integration/ip/config/{ParserUnitTests-context.xml => ParserTests-context.xml} (98%) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index d374fbed90..8adaacd46b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -131,6 +131,8 @@ public abstract class IpAdapterParserUtils { public static final String SSL_HANDSHAKE_TIMEOUT = "ssl-handshake-timeout"; + public static final String CONNECT_TIMEOUT = "connect-timeout"; + private IpAdapterParserUtils() { } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java index 0748b22cc0..3e119f04db 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java @@ -59,73 +59,75 @@ import org.springframework.util.Assert; public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean implements Lifecycle, BeanNameAware, ApplicationEventPublisherAware { - private volatile AbstractConnectionFactory connectionFactory; + private AbstractConnectionFactory connectionFactory; - private volatile String type; + private String type; - private volatile String host; + private String host; - private volatile int port; + private int port; - private volatile int soTimeout; + private int soTimeout; - private volatile int soSendBufferSize; + private int soSendBufferSize; - private volatile int soReceiveBufferSize; + private int soReceiveBufferSize; - private volatile boolean soTcpNoDelay; + private boolean soTcpNoDelay; - private volatile int soLinger = -1; // don't set by default + private int soLinger = -1; // don't set by default - private volatile boolean soKeepAlive; + private boolean soKeepAlive; - private volatile int soTrafficClass = -1; // don't set by default + private int soTrafficClass = -1; // don't set by default - private volatile Executor taskExecutor; + private Executor taskExecutor; - private volatile Deserializer deserializer = new ByteArrayCrLfSerializer(); + private Deserializer deserializer = new ByteArrayCrLfSerializer(); - private volatile Serializer serializer = new ByteArrayCrLfSerializer(); + private Serializer serializer = new ByteArrayCrLfSerializer(); - private volatile TcpMessageMapper mapper = new TcpMessageMapper(); + private TcpMessageMapper mapper = new TcpMessageMapper(); - private volatile boolean mapperSet; + private boolean mapperSet; - private volatile boolean singleUse; + private boolean singleUse; - private volatile int backlog = 5; + private int backlog = 5; - private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain; + private TcpConnectionInterceptorFactoryChain interceptorFactoryChain; - private volatile boolean lookupHost = true; + private boolean lookupHost = true; - private volatile String localAddress; + private String localAddress; - private volatile boolean usingNio; + private boolean usingNio; - private volatile boolean usingDirectBuffers; + private boolean usingDirectBuffers; - private volatile String beanName; + private String beanName; - private volatile boolean applySequence; + private boolean applySequence; - private volatile Long readDelay; + private Long readDelay; - private volatile TcpSSLContextSupport sslContextSupport; + private TcpSSLContextSupport sslContextSupport; - private volatile Integer sslHandshakeTimeout; + private Integer sslHandshakeTimeout; - private volatile TcpSocketSupport socketSupport = new DefaultTcpSocketSupport(); + private TcpSocketSupport socketSupport = new DefaultTcpSocketSupport(); - private volatile TcpNioConnectionSupport nioConnectionSupport; + private TcpNioConnectionSupport nioConnectionSupport; - private volatile TcpNetConnectionSupport netConnectionSupport; + private TcpNetConnectionSupport netConnectionSupport; - private volatile TcpSocketFactorySupport socketFactorySupport; + private TcpSocketFactorySupport socketFactorySupport; - private volatile ApplicationEventPublisher applicationEventPublisher; + private ApplicationEventPublisher applicationEventPublisher; - private volatile BeanFactory beanFactory; + private BeanFactory beanFactory; + + private Integer connectTimeout; public TcpConnectionFactoryFactoryBean() { @@ -189,6 +191,9 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean getComponentsToRegister() { return this.connectionFactory != null diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index 4daa96f042..e43269dbf7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp; +import java.io.IOException; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -84,6 +85,8 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler private int secondChanceDelay = DEFAULT_SECOND_CHANCE_DELAY; + private boolean closeStreamAfterSend; + /** * @param requestTimeout the requestTimeout to set */ @@ -117,6 +120,8 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler if (!this.evaluationContextSet) { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); } + Assert.state(!this.closeStreamAfterSend || this.isSingleUse, + "Single use connection needed with closeStreamAfterSend"); } /** @@ -149,9 +154,12 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler logger.debug("Added pending reply " + connectionId); } connection.send(requestMessage); + if (this.closeStreamAfterSend) { + connection.shutdownOutput(); + } return getReply(requestMessage, connection, connectionId, reply); } - catch (RuntimeException e) { + catch (RuntimeException | IOException e) { logger.error("Tcp Gateway exception", e); if (e instanceof MessagingException) { throw (MessagingException) e; @@ -305,6 +313,18 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler this.setOutputChannelName(replyChannel); } + /** + * Set to true to close the connection ouput stream after sending without + * closing the connection. Use to signal EOF to the server, such as when using + * a {@link org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer}. + * Requires a single-use connection factory. + * @param closeStreamAfterSend true to close. + * @since 5.2 + */ + public void setCloseStreamAfterSend(boolean closeStreamAfterSend) { + this.closeStreamAfterSend = closeStreamAfterSend; + } + @Override public String getComponentType() { return "ip:tcp-outbound-gateway"; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 85a7f5685e..1114fef9b6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp.connection; import java.net.Socket; +import java.time.Duration; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -33,11 +34,15 @@ import org.springframework.lang.Nullable; */ public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory { + private static final long DEFAULT_CONNECT_TIMEOUT = 60L; + private final ReadWriteLock theConnectionLock = new ReentrantReadWriteLock(); - private volatile TcpConnectionSupport theConnection; + private boolean manualListenerRegistration; - private volatile boolean manualListenerRegistration; + private Duration connectTimeout = Duration.ofSeconds(DEFAULT_CONNECT_TIMEOUT); + + private volatile TcpConnectionSupport theConnection; /** * Constructs a factory that will established connections to the host and port. @@ -48,6 +53,19 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection super(host, port); } + /** + * Set the connection timeout in seconds. Defaults to 60. + * @param connectTimeout the timeout. + * @since 5.2 + */ + public void setConnectTimeout(int connectTimeout) { + this.connectTimeout = Duration.ofSeconds(connectTimeout); + } + + protected Duration getConnectTimeout() { + return this.connectTimeout; + } + /** * Set whether to automatically (default) or manually add a {@link TcpListener} to the * connections created by this factory. By default, the factory automatically configures diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index 06fe3292f7..5712207837 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -16,6 +16,8 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.IOException; + import javax.net.ssl.SSLSession; import org.springframework.core.serializer.Deserializer; @@ -132,4 +134,24 @@ public interface TcpConnection extends Runnable { */ SocketInfo getSocketInfo(); + /** + * Set the connection's input stream to end of stream. + * @throws IOException an IO Exception. + * @since 5.2 + */ + @SuppressWarnings("unused") + default void shutdownInput() throws IOException { + throw new UnsupportedOperationException("This connection does not support shutDownInput()"); + } + + /** + * Disable the socket's output stream. + * @throws IOException an IO Exception + * @since 5.2 + */ + @SuppressWarnings("unused") + default void shutdownOutput() throws IOException { + throw new UnsupportedOperationException("This connection does not support shutDownOutput()"); + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index e48edd316b..eaf8976d3a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp.connection; import java.io.IOException; import java.io.UncheckedIOException; +import java.net.InetSocketAddress; import java.net.Socket; import org.springframework.util.Assert; @@ -88,7 +89,9 @@ public class TcpNetClientConnectionFactory extends * @throws IOException Any IOException. */ protected Socket createSocket(String host, int port) throws IOException { - return this.tcpSocketFactorySupport.getSocketFactory().createSocket(host, port); + Socket socket = this.tcpSocketFactorySupport.getSocketFactory().createSocket(); + socket.connect(new InetSocketAddress(host, port), (int) getConnectTimeout().toMillis()); + return socket; } protected TcpSocketFactorySupport getTcpSocketFactorySupport() { 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 cdbc5da628..27c677380f 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 @@ -285,4 +285,24 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling } } + /** + * Set the socket's input stream to end of stream. + * @throws IOException an IO Exception. + * @since 5.2 + * @see Socket#shutdownInput() + */ + public void shutdownInput() throws IOException { + this.socket.shutdownInput(); + } + + /** + * Disable the socket's output stream. + * @throws IOException an IO Exception + * @since 5.2 + * @see Socket#shutdownOutput() + */ + public void shutdownOutput() throws IOException { + this.socket.shutdownOutput(); + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index ab365f08c5..03d7e42365 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -85,7 +85,7 @@ public class TcpNioClientConnectionFactory extends @Override protected TcpConnectionSupport buildNewConnection() { try { - SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(getHost(), getPort())); + SocketChannel socketChannel = SocketChannel.open(); setSocketAttributes(socketChannel.socket()); TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(socketChannel, false, isLookupHost(), @@ -99,6 +99,17 @@ public class TcpNioClientConnectionFactory extends TcpConnectionSupport wrappedConnection = wrapConnection(connection); initializeConnection(wrappedConnection, socketChannel.socket()); socketChannel.configureBlocking(false); + socketChannel.connect(new InetSocketAddress(getHost(), getPort())); + boolean connected = socketChannel.finishConnect(); + long timeLeft = getConnectTimeout().toMillis(); + while (!connected && timeLeft > 0) { + Thread.sleep(50); // NOSONAR Magic # + connected = socketChannel.finishConnect(); + timeLeft -= 50; // NOSONAR Magic # + } + if (!connected) { + throw new IOException("Not connected after connectTimeout"); + } if (getSoTimeout() > 0) { connection.setLastRead(System.currentTimeMillis()); } @@ -110,6 +121,10 @@ public class TcpNioClientConnectionFactory extends catch (IOException e) { throw new UncheckedIOException(e); } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new UncheckedIOException(new IOException(e)); + } } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index a8f4ce8389..bc4ddbb76a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -574,6 +574,26 @@ public class TcpNioConnection extends TcpConnectionSupport { return this.lastSend; } + /** + * Set the socket's input stream to end of stream. + * @throws IOException an IO Exception. + * @since 5.2 + * @see SocketChannel#shutdownInput() + */ + public void shutdownInput() throws IOException { + this.socketChannel.shutdownInput(); + } + + /** + * Disable the socket's output stream. + * @throws IOException an IO Exception + * @since 5.2 + * @see SocketChannel#shutdownOutput() + */ + public void shutdownOutput() throws IOException { + this.socketChannel.shutdownOutput(); + } + /** * OutputStream to wrap a SocketChannel; implements timeout on write. * diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.2.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.2.xsd index aee2716660..c63ddf555a 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.2.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.2.xsd @@ -478,6 +478,15 @@ + + + + Close the output stream after sending the message; this signals + EOF to the server while keeping the connection open to receive + the reply. Requires 'single-use' set to 'true'. + + + @@ -792,6 +801,14 @@ + + + + For client factories, the amount of time to wait for a connection to + be established. + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml similarity index 98% rename from spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml rename to spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml index 624a03c5ee..a89593582a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml @@ -179,6 +179,7 @@ host="localhost" lookup-host="false" apply-sequence="false" + connect-timeout="70" read-delay="10000" /> @@ -190,10 +191,6 @@ phase="125" /> - - - - @@ -261,6 +258,7 @@ request-channel="tcpAdviceGateChannel" reply-channel="replyChannel" remote-timeout-expression="4000" + close-stream-after-send="true" connection-factory="mockClientCf"> diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 1cbee5e57f..5d8bb79d17 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -17,19 +17,24 @@ package org.springframework.integration.ip.config; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import java.time.Duration; import java.util.Iterator; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.ImportResource; import org.springframework.core.io.UrlResource; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -42,6 +47,7 @@ import org.springframework.integration.ip.tcp.TcpInboundGateway; import org.springframework.integration.ip.tcp.TcpOutboundGateway; import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.DefaultTcpNetConnectionSupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport; @@ -69,8 +75,7 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell @@ -79,8 +84,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * * @since 2.0 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class ParserUnitTests { @@ -427,6 +431,7 @@ public class ParserUnitTests { assertThat((Boolean) TestUtils.getPropertyValue( TestUtils.getPropertyValue(cfC1, "mapper"), "applySequence")).isFalse(); assertThat(TestUtils.getPropertyValue(cfC1, "readDelay")).isEqualTo(10000L); + assertThat(TestUtils.getPropertyValue(cfC1, "connectTimeout")).isEqualTo(Duration.ofSeconds(70)); } @Test @@ -476,6 +481,7 @@ public class ParserUnitTests { assertThat(TestUtils.getPropertyValue(outAdviceGateway, "remoteTimeoutExpression.expression")) .isEqualTo("4000"); + assertThat(TestUtils.getPropertyValue(outAdviceGateway, "closeStreamAfterSend")).isEqualTo(Boolean.TRUE); } @Test @@ -675,4 +681,18 @@ public class ParserUnitTests { super(connection, connectionFactoryName); } } + + @Configuration + @ImportResource("org/springframework/integration/ip/config/ParserTests-context.xml") + public static class Config { + + @Bean + AbstractClientConnectionFactory mockClientCf() { + AbstractClientConnectionFactory mock = mock(AbstractClientConnectionFactory.class); + given(mock.isSingleUse()).willReturn(true); + return mock; + } + + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java index d86447b6d7..4b9dc2da53 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.intercept.MethodInterceptor; import org.junit.Test; @@ -30,6 +31,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationListener; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.QueueChannel; @@ -37,13 +39,17 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageChannels; import org.springframework.integration.dsl.Transformers; import org.springframework.integration.dsl.context.IntegrationFlowContext; +import org.springframework.integration.dsl.context.IntegrationFlowContext.IntegrationFlowRegistration; import org.springframework.integration.ip.tcp.TcpOutboundGateway; import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionServerListeningEvent; +import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer; import org.springframework.integration.ip.tcp.serializer.TcpCodecs; import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; import org.springframework.integration.ip.udp.UdpServerListeningEvent; @@ -67,6 +73,9 @@ import org.springframework.test.context.junit4.SpringRunner; @DirtiesContext public class IpIntegrationTests { + @Autowired + private ConfigurableApplicationContext applicationContext; + @Autowired private AbstractServerConnectionFactory server1; @@ -164,6 +173,42 @@ public class IpIntegrationTests { assertThat(udpMulticastOutboundChannelAdapterSpec2.get()).isInstanceOf(MulticastSendingMessageHandler.class); } + @Test + public void testCloseStream() throws InterruptedException { + IntegrationFlow server = IntegrationFlows.from(Tcp.inboundGateway(Tcp.netServer(0) + .deserializer(new ByteArrayRawSerializer()))) + .transform(p -> "reply:" + new String(p).toUpperCase()) + .get(); + CountDownLatch latch = new CountDownLatch(1); + AtomicInteger port = new AtomicInteger(); + class Listener implements ApplicationListener { + + @Override + public void onApplicationEvent(TcpConnectionServerListeningEvent event) { + port.set(event.getPort()); + latch.countDown(); + } + + } + this.applicationContext.addApplicationListener(new Listener()); + this.flowContext.registration(server) + .id("streamCloseServer") + .register(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + IntegrationFlow client = IntegrationFlows.from(MessageChannels.direct()) + .handle(Tcp.outboundGateway(Tcp.netClient("localhost", port.get()) + .singleUseConnections(true) + .serializer(new ByteArrayRawSerializer())) + .closeStreamAfterSend(true)) + .transform(Transformers.objectToString()) + .get(); + IntegrationFlowRegistration clientRegistration = this.flowContext.registration(client) + .id("streamCloseClient") + .register(); + assertThat(clientRegistration.getMessagingTemplate() + .convertSendAndReceive("foo", String.class)).isEqualTo("reply:FOO"); + } + @Configuration @EnableIntegration public static class Config { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index a0fc8d06c4..532bfd08e1 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -31,6 +31,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; @@ -38,14 +39,20 @@ import javax.net.SocketFactory; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; +import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -284,6 +291,57 @@ public class TcpInboundGatewayTests { scf.stop(); } + @Test + public void testNetCloseStream() throws InterruptedException, IOException { + testCloseStream(new TcpNetServerConnectionFactory(0), + port -> new TcpNetClientConnectionFactory("localhost", port)); + } + + @Test + public void testNioCloseStream() throws InterruptedException, IOException { + testCloseStream(new TcpNioServerConnectionFactory(0), + port -> new TcpNioClientConnectionFactory("localhost", port)); + } + + private void testCloseStream(AbstractServerConnectionFactory scf, + Function ccf) throws InterruptedException, IOException { + + scf.setSingleUse(true); + scf.setDeserializer(new ByteArrayRawSerializer()); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(scf); + BeanFactory bf = mock(ConfigurableBeanFactory.class); + gateway.setBeanFactory(bf); + gateway.start(); + TestingUtilities.waitListening(scf, 20000L); + int port = scf.getPort(); + final DirectChannel channel = new DirectChannel(); + gateway.setRequestChannel(channel); + BridgeHandler bridge = new BridgeHandler(); + bridge.setBeanFactory(bf); + bridge.afterPropertiesSet(); + ConsumerEndpointFactoryBean consumer = new ConsumerEndpointFactoryBean(); + consumer.setInputChannel(channel); + consumer.setBeanFactory(bf); + consumer.setHandler(bridge); + consumer.afterPropertiesSet(); + consumer.start(); + AbstractClientConnectionFactory client = ccf.apply(port); + CountDownLatch latch = new CountDownLatch(1); + client.registerListener(message -> { + latch.countDown(); + return false; + }); + client.afterPropertiesSet(); + client.start(); + TcpConnectionSupport connection = client.getConnection(); + connection.send(new GenericMessage<>("foo")); + connection.shutdownOutput(); // signal EOF to server + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + gateway.stop(); + client.stop(); + } + private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java index a77b2d07fc..770ed74ff7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java @@ -18,20 +18,29 @@ package org.springframework.integration.ip.tcp.connection; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.InputStream; +import java.io.UncheckedIOException; import java.net.InetAddress; import java.net.ServerSocket; import java.net.Socket; +import java.net.SocketException; +import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -69,7 +78,7 @@ public class SocketSupportTests { when(socket.getInputStream()).thenReturn(is); InetAddress inetAddress = InetAddress.getLocalHost(); when(socket.getInetAddress()).thenReturn(inetAddress); - when(factory.createSocket("x", 0)).thenReturn(socket); + when(factory.createSocket()).thenReturn(socket); TcpSocketSupport socketSupport = Mockito.mock(TcpSocketSupport.class); TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory("x", 0); @@ -83,25 +92,64 @@ public class SocketSupportTests { } @Test - public void testNetServer() throws Exception { + public void testNetClientSocketTimeout() throws Exception { TcpSocketFactorySupport factorySupport = mock(TcpSocketFactorySupport.class); - ServerSocketFactory factory = mock(ServerSocketFactory.class); - when(factorySupport.getServerSocketFactory()).thenReturn(factory); + SocketFactory factory = Mockito.mock(SocketFactory.class); + when(factorySupport.getSocketFactory()).thenReturn(factory); Socket socket = mock(Socket.class); InputStream is = mock(InputStream.class); when(is.read()).thenReturn(-1); when(socket.getInputStream()).thenReturn(is); InetAddress inetAddress = InetAddress.getLocalHost(); when(socket.getInetAddress()).thenReturn(inetAddress); + when(factory.createSocket()).thenReturn(socket); + doThrow(new SocketTimeoutException()).when(socket).connect(any(), eq(1000)); + TcpSocketSupport socketSupport = Mockito.mock(TcpSocketSupport.class); + + TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory("x", 0); + connectionFactory.setConnectTimeout(1); + connectionFactory.setTcpSocketFactorySupport(factorySupport); + connectionFactory.setTcpSocketSupport(socketSupport); + connectionFactory.start(); + assertThatThrownBy(() -> connectionFactory.getConnection()) + .isInstanceOf(UncheckedIOException.class) + .hasCauseInstanceOf(SocketTimeoutException.class); + + connectionFactory.stop(); + } + + @Test + public void testNetServer() throws Exception { + TcpSocketFactorySupport factorySupport = mock(TcpSocketFactorySupport.class); + ServerSocketFactory factory = mock(ServerSocketFactory.class); + when(factorySupport.getServerSocketFactory()).thenReturn(factory); + Socket socket = mock(Socket.class); + Socket socket1 = mock(Socket.class); + InputStream is = mock(InputStream.class); + when(is.read()).thenReturn(-1); + when(socket.getInputStream()).thenReturn(is); + when(socket1.getInputStream()).thenReturn(is); + InetAddress inetAddress = InetAddress.getLocalHost(); + when(socket.getInetAddress()).thenReturn(inetAddress); + when(socket1.getInetAddress()).thenReturn(inetAddress); ServerSocket serverSocket = mock(ServerSocket.class); + AtomicBoolean closed = new AtomicBoolean(); + doAnswer(invoc -> { + closed.set(true); + return null; + }).when(serverSocket).close(); when(serverSocket.getInetAddress()).thenReturn(inetAddress); when(factory.createServerSocket(0, 5)).thenReturn(serverSocket); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); when(serverSocket.accept()).thenReturn(socket).then(invocation -> { + if (closed.get()) { + throw new SocketException(); + } latch1.countDown(); latch2.await(10, TimeUnit.SECONDS); - return null; + Thread.sleep(50); + return socket1; }); TcpSocketSupport socketSupport = mock(TcpSocketSupport.class); diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 1b94d1666e..6b1a655463 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -410,6 +410,7 @@ Doing so causes the adapter to close the socket after sending the message. The serializer does not, by itself, close the connection. You should use this serializer only with the connection factories used by channel adapters (not gateways), and the connection factories should be used by either an inbound or outbound adapter but not both. See also `ByteArrayElasticRawDeserializer`, later in this section. +However, since version 5.2, the outbound gateway has a new property `closeStreamAfterSend`; this allows the use of raw serializers/deserializers because the EOF is signaled to the server, while leaving the connection open to receive the reply. NOTE: Before version 4.2.2, when using non-blocking I/O (NIO), this serializer treated a timeout (during read) as an end of file, and the data read so far was emitted as a message. This is unreliable and should not be used to delimit messages. @@ -814,6 +815,10 @@ remote-timeout-expression="headers['timeout']" --> `client-mode` is not currently available with the outbound gateway. +Starting with version 5.2, the outbound gateway can be configured with the property `closeStreamAfterSend`. +If the connection factory is configured for `single-use` (a new connection for each request/reply) the gateway will close the output stream; this signals EOF to the server. +This is useful if the server uses the EOF to determine the end of message, rather than some delimiter in the stream, but leaves the connection open in order to receive the reply. + [[ip-correlation]] === TCP Message Correlation diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 61ff1ff1da..9115ab9364 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -69,6 +69,13 @@ See <<./ip.adoc#tcp-codecs,Message Demarcation (Serializers and Deserializers)>> When using a `TcpNioServerConnectionFactory`, priority is now given to accepting new connections over reading from existing connections, but it is configurable. See <<./ip.adoc#note-nio,About Non-blocking I/O (NIO)>> for more information. +The outbound gateway has a new property `closeStreamAfterSend`; when used with a new connection for each request/reply it signals EOF to the server, without closing the connection. +This is useful for servers that use the EOF to signal end of message instead of some delimiter in the data. +See <> 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 <> for more information. + [[x5.2-mail]] ==== Mail Changes