Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

TCP: Connect Timeout; Close Stream

Browse Source 
- Add `connectTimeout` to client connection factories
- Add `closeStreamAfterSend` to outbound gateway

* Polishing - PR Comments.
This commit is contained in:
Gary Russell
2019-06-05 14:38:14 -04:00
committed by Artem Bilan
parent cd075723e2
commit 466daa8774
22 changed files with 420 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

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

77
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -59,73 +59,75 @@ import org.springframework.util.Assert;
public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory>
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<Abstrac
this.setCommonAttributes(factory);
factory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
factory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
if (this.connectTimeout != null) {
factory.setConnectTimeout(this.connectTimeout);
}
this.connectionFactory = factory;
}
}
@@ -501,6 +506,10 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.applicationEventPublisher = applicationEventPublisher;
}
public void setConnectTimeout(int connectTimeout) {
this.connectTimeout = connectTimeout;
}
/**
* Set the SSL handshake timeout (only used with SSL and NIO).
* @param sslHandshakeTimeout the timeout.

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryParser.java
View File

@@ -100,6 +100,8 @@ public class TcpConnectionFactoryParser extends AbstractBeanDefinitionParser {
IpAdapterParserUtils.SSL_HANDSHAKE_TIMEOUT);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.READ_DELAY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CONNECT_TIMEOUT);
return builder.getBeanDefinition();
}

1
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
View File

@@ -55,6 +55,7 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.REPLY_TIMEOUT, "sendTimeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "close-stream-after-send");
return builder;
}

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractConnectionFactorySpec.java
View File

@@ -219,4 +219,15 @@ public abstract class AbstractConnectionFactorySpec
return _this();
}
/**
* This connection factory uses a new connection for each operation.
* @param single true for a new connection for each operation.
* @return the spec.
* @since 5.2
*/
public S singleUseConnections(boolean single) {
this.target.setSingleUse(single);
return _this();
}
}

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpClientConnectionFactorySpec.java
View File

@@ -38,4 +38,15 @@ public class TcpClientConnectionFactorySpec
super(nio ? new TcpNioClientConnectionFactory(host, port) : new TcpNetClientConnectionFactory(host, port));
}
/**
* Set the connection timeout in seconds. Defaults to 60.
* @param connectTimeout the timeout.
* @return the spec.
* @since 5.2
*/
public TcpClientConnectionFactorySpec connectTimeout(int connectTimeout) {
this.target.setConnectTimeout(connectTimeout);
return _this();
}
}

14
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java
View File

@@ -89,6 +89,20 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
return _this();
}
/**
* 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.
* @return the spec.
* @since 5.2
*/
public TcpOutboundGatewaySpec closeStreamAfterSend(boolean closeStreamAfterSend) {
this.target.setCloseStreamAfterSend(closeStreamAfterSend);
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return this.connectionFactory != null

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

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

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java
View File

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

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java
View File

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

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java
View File

@@ -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() {

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

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

17
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

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

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

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

17
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.2.xsd
View File

@@ -478,6 +478,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="close-stream-after-send" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup" />
</xsd:complexType>
</xsd:element>
@@ -792,6 +801,14 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="connect-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
For client factories, the amount of time to wait for a connection to
be established.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

6
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml → spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml
View File

@@ -179,6 +179,7 @@
host="localhost"
lookup-host="false"
apply-sequence="false"
connect-timeout="70"
read-delay="10000"
/>
@@ -190,10 +191,6 @@
phase="125"
/>
<bean id="mockClientCf" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory" />
</bean>
<int:channel id="tcpAdviceChannel">
<int:queue/>
</int:channel>
@@ -261,6 +258,7 @@
request-channel="tcpAdviceGateChannel"
reply-channel="replyChannel"
remote-timeout-expression="4000"
close-stream-after-send="true"
connection-factory="mockClientCf">
<int:poller fixed-delay="100"/>
<ip:request-handler-advice-chain>

32
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

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

45
spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java
View File

@@ -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())))
.<byte[], String>transform(p -> "reply:" + new String(p).toUpperCase())
.get();
CountDownLatch latch = new CountDownLatch(1);
AtomicInteger port = new AtomicInteger();
class Listener implements ApplicationListener<TcpConnectionServerListeningEvent> {
@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 {

58
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java
View File

@@ -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<Integer, AbstractClientConnectionFactory> 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++) {

58
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java
View File

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

5
src/reference/asciidoc/ip.adoc
View File

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

7
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<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 <<tcp-connection-factory>> for more information.
[[x5.2-mail]]
==== Mail Changes