diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java index e4229807b3..bb2af7ec63 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -56,6 +56,7 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.REPLY_TIMEOUT, "sendTimeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "close-stream-after-send"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async"); return builder; } 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 6b85bca77b..1963f0d591 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 @@ -1,5 +1,5 @@ /* - * Copyright 2001-2019 the original author or authors. + * Copyright 2001-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.util.Date; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -38,7 +39,9 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpConnectionFailedCorrelationEvent; +import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.ip.tcp.connection.TcpNioConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpSender; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -46,6 +49,7 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; +import org.springframework.util.concurrent.SettableListenableFuture; /** * TCP outbound gateway that uses a client connection factory. If the factory is configured @@ -123,6 +127,21 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler } Assert.state(!this.closeStreamAfterSend || this.isSingleUse, "Single use connection needed with closeStreamAfterSend"); + if (isAsync()) { + try { + TcpConnectionSupport connection = this.connectionFactory.getConnection(); + if (connection instanceof TcpNioConnectionSupport) { + setAsync(false); + this.logger.warn("Async replies are not supported with NIO; see the reference manual"); + } + if (this.isSingleUse) { + connection.close(); + } + } + catch (Exception e) { + this.logger.error("Could not check if async is supported", e); + } + } } /** @@ -144,11 +163,12 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler boolean haveSemaphore = false; TcpConnection connection = null; String connectionId = null; + boolean async = isAsync(); try { haveSemaphore = acquireSemaphoreIfNeeded(requestMessage); connection = this.connectionFactory.getConnection(); Long remoteTimeout = getRemoteTimeout(requestMessage); - AsyncReply reply = new AsyncReply(remoteTimeout); + AsyncReply reply = new AsyncReply(remoteTimeout, connection, haveSemaphore, requestMessage, async); connectionId = connection.getConnectionId(); this.pendingReplies.put(connectionId, reply); if (logger.isDebugEnabled()) { @@ -158,7 +178,12 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler if (this.closeStreamAfterSend) { connection.shutdownOutput(); } - return getReply(requestMessage, connection, connectionId, reply); + if (async) { + return reply.getFuture(); + } + else { + return getReply(requestMessage, connection, connectionId, reply); + } } catch (RuntimeException | IOException e) { logger.error("Tcp Gateway exception", e); @@ -172,7 +197,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler throw new MessageHandlingException(requestMessage, "Interrupted in the [" + this + ']', e); } finally { - cleanUp(haveSemaphore, connection, connectionId); + if (!async) { + cleanUp(haveSemaphore, connection, connectionId); + } } } @@ -264,7 +291,13 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler return false; } } - reply.setReply(message); + if (isAsync()) { + reply.getFuture().set(message); + cleanUp(reply.isHaveSemaphore(), reply.getConnection(), connectionId); + } + else { + reply.setReply(message); + } return false; } @@ -365,19 +398,44 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler private final long remoteTimeout; + private final TcpConnection connection; + + private final boolean haveSemaphore; + + private final SettableListenableFuture> future = new SettableListenableFuture<>(); + private volatile Message reply; - private AsyncReply(long remoteTimeout) { + AsyncReply(long remoteTimeout, TcpConnection connection, boolean haveSemaphore, Message requestMessage, + boolean async) { + this.latch = new CountDownLatch(1); this.secondChanceLatch = new CountDownLatch(1); this.remoteTimeout = remoteTimeout; + this.connection = connection; + this.haveSemaphore = haveSemaphore; + if (async && remoteTimeout > 0) { + getTaskScheduler().schedule(() -> { + TcpOutboundGateway.this.pendingReplies.remove(connection.getConnectionId()); + this.future.setException( + new MessageTimeoutException(requestMessage, "Timed out waiting for response")); + }, new Date(System.currentTimeMillis() + remoteTimeout)); + } + } + + TcpConnection getConnection() { + return this.connection; + } + + boolean isHaveSemaphore() { + return this.haveSemaphore; } /** * Sender blocks here until the reply is received, or we time out * @return The return message or null if we time out */ - public Message getReply() { + Message getReply() { try { if (!this.latch.await(this.remoteTimeout, TimeUnit.MILLISECONDS)) { return null; @@ -411,6 +469,10 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler return this.reply; } + SettableListenableFuture> getFuture() { + return this.future; + } + private void doThrowErrorMessagePayload() { if (this.reply.getPayload() instanceof MessagingException) { throw (MessagingException) this.reply.getPayload(); diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd index 96768d49c6..3c033e0155 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd @@ -464,7 +464,7 @@ - Specifies an expresssion that is evaluated against the outbound message + Specifies an expression that is evaluated against the outbound message to determine the time the gateway will wait for a reply from the remote system. Mutually exclusive with 'remote-timeout'. @@ -488,6 +488,16 @@ + + + + Set to true for async request/reply - see reference manual. + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml index a89593582a..0688fe2d5b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml @@ -248,6 +248,7 @@ order="24" auto-startup="false" phase="127" + async="true" /> 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 5d8bb79d17..d71b917c91 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -478,10 +478,12 @@ public class ParserUnitTests { assertThat(tcpOutboundGateway.getComponentType()).isEqualTo("ip:tcp-outbound-gateway"); assertThat(cfC2.isLookupHost()).isTrue(); assertThat(dfa.getPropertyValue("order")).isEqualTo(24); + assertThat(dfa.getPropertyValue("async")).isEqualTo(Boolean.TRUE); assertThat(TestUtils.getPropertyValue(outAdviceGateway, "remoteTimeoutExpression.expression")) .isEqualTo("4000"); assertThat(TestUtils.getPropertyValue(outAdviceGateway, "closeStreamAfterSend")).isEqualTo(Boolean.TRUE); + assertThat(TestUtils.getPropertyValue(outAdviceGateway, "async")).isEqualTo(Boolean.FALSE); } @Test diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index d8ccc01ae0..3860b2b932 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,8 +26,10 @@ import static org.mockito.Mockito.when; import java.io.EOFException; import java.io.IOException; +import java.io.InputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; +import java.io.OutputStream; import java.io.UncheckedIOException; import java.net.ServerSocket; import java.net.Socket; @@ -49,9 +51,7 @@ import javax.net.ServerSocketFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.junit.ClassRule; -import org.junit.Rule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; @@ -71,16 +71,20 @@ import org.springframework.integration.ip.tcp.connection.FailoverClientConnectio import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; +import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; +import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.rule.Log4j2LevelAdjuster; -import org.springframework.integration.test.support.LongRunningIntegrationTest; +import org.springframework.integration.test.condition.LongRunningTest; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell @@ -88,21 +92,15 @@ import org.springframework.messaging.support.GenericMessage; * * @since 2.0 */ +@LongRunningTest public class TcpOutboundGatewayTests { private static final Log logger = LogFactory.getLog(TcpOutboundGatewayTests.class); private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); - @ClassRule - public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest(); - - @Rule - public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace(); - - @Test - public void testGoodNetSingle() throws Exception { + void testGoodNetSingle() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final AtomicReference serverSocket = new AtomicReference<>(); @@ -167,7 +165,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testGoodNetMultiplex() throws Exception { + void testGoodNetMultiplex() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final AtomicReference serverSocket = new AtomicReference<>(); @@ -223,7 +221,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testGoodNetTimeout() throws Exception { + void testGoodNetTimeout() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); final AtomicReference serverSocket = new AtomicReference<>(); @@ -304,7 +302,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testGoodNetGWTimeout() throws Exception { + void testGoodNetGWTimeout() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = buildCF(port); @@ -314,7 +312,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testGoodNetGWTimeoutCached() throws Exception { + void testGoodNetGWTimeoutCached() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = buildCF(port); @@ -446,7 +444,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testCachingFailover() throws Exception { + void testCachingFailover() throws Exception { final AtomicReference serverSocket = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -528,7 +526,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testFailoverCached() throws Exception { + void testFailoverCached() throws Exception { final AtomicReference serverSocket = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -619,7 +617,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNetGWPropagatesSocketClose() throws Exception { + void testNetGWPropagatesSocketClose() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); @@ -633,7 +631,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNioGWPropagatesSocketClose() throws Exception { + void testNioGWPropagatesSocketClose() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); @@ -647,7 +645,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testCachedGWPropagatesSocketClose() throws Exception { + void testCachedGWPropagatesSocketClose() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); @@ -662,7 +660,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testFailoverGWPropagatesSocketClose() throws Exception { + void testFailoverGWPropagatesSocketClose() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); @@ -746,7 +744,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNetGWPropagatesSocketTimeout() throws Exception { + void testNetGWPropagatesSocketTimeout() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); @@ -760,7 +758,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNioGWPropagatesSocketTimeout() throws Exception { + void testNioGWPropagatesSocketTimeout() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); @@ -774,7 +772,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNetGWPropagatesSocketTimeoutSingleUse() throws Exception { + void testNetGWPropagatesSocketTimeoutSingleUse() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); @@ -788,7 +786,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNioGWPropagatesSocketTimeoutSingleUse() throws Exception { + void testNioGWPropagatesSocketTimeoutSingleUse() throws Exception { ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); final int port = serverSocket.getLocalPort(); AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); @@ -853,7 +851,7 @@ public class TcpOutboundGatewayTests { } @Test - public void testNioSecondChance() throws Exception { + void testNioSecondChance() throws Exception { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); final int port = server.getLocalPort(); TcpOutboundGateway gateway = new TcpOutboundGateway(); @@ -908,4 +906,164 @@ public class TcpOutboundGatewayTests { server.close(); } + @Test + void testAsyncSingle() throws Exception { + testAsync(true); + } + + @Test + void testAsyncShared() throws Exception { + testAsync(false); + } + + private void testAsync(boolean singleUse) throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference serverSocket = new AtomicReference<>(); + ThreadPoolTaskScheduler sched = new ThreadPoolTaskScheduler(); + sched.initialize(); + TcpOutboundGateway gateway = null; + try { + this.executor.execute(() -> { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100); + serverSocket.set(server); + latch.countDown(); + int i = 0; + while (true) { + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + OutputStream os = socket.getOutputStream(); + ByteArrayCrLfSerializer deser = new ByteArrayCrLfSerializer(); + try { + deser.deserialize(is); + } + catch (SoftEndOfStreamException e) { + continue; + } + deser.serialize(("reply" + ++i).getBytes(), os); + if (!singleUse) { + deser.deserialize(is); + deser.serialize(("reply" + ++i).getBytes(), os); + } + socket.close(); + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + }); + assertThat(latch.await(10000, TimeUnit.MILLISECONDS)).isTrue(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(singleUse); + ccf.start(); + gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(ccf); + gateway.setAsync(true); + QueueChannel replyChannel = new QueueChannel(); + AtomicReference thread = new AtomicReference<>(); + replyChannel.addInterceptor(new ChannelInterceptor() { + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + thread.set(Thread.currentThread()); + } + + }); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setTaskScheduler(sched); + gateway.afterPropertiesSet(); + gateway.handleMessage(MessageBuilder.withPayload("Test1").build()); + gateway.handleMessage(MessageBuilder.withPayload("Test2").build()); + Message reply = replyChannel.receive(10000); + assertThat(reply).isNotNull(); + assertThat(reply.getPayload()).isEqualTo("reply1".getBytes()); + reply = replyChannel.receive(10000); + assertThat(reply).isNotNull(); + assertThat(reply.getPayload()).isEqualTo("reply2".getBytes()); + assertThat(thread.get()).isNotSameAs(Thread.currentThread()); + } + finally { + if (gateway != null) { + gateway.stop(); + } + done.set(true); + if (serverSocket.get() != null) { + serverSocket.get().close(); + } + sched.shutdown(); + } + } + + @Test + void testAsyncTimeout() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch doneLatch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference serverSocket = new AtomicReference<>(); + AbstractClientConnectionFactory ccf = null; + ThreadPoolTaskScheduler sched = new ThreadPoolTaskScheduler(); + sched.initialize(); + try { + this.executor.execute(() -> { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100); + serverSocket.set(server); + latch.countDown(); + int i = 0; + while (true) { + Socket socket = server.accept(); + doneLatch.await(10, TimeUnit.SECONDS); + socket.close(); + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + }); + assertThat(latch.await(10000, TimeUnit.MILLISECONDS)).isTrue(); + ccf = new TcpNetClientConnectionFactory("localhost", + serverSocket.get().getLocalPort()); + ccf.setSoTimeout(10000); + ccf.start(); + TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(ccf); + gateway.setAsync(true); + gateway.setRemoteTimeout(10); + QueueChannel replyChannel = new QueueChannel(); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setTaskScheduler(sched); + gateway.afterPropertiesSet(); + QueueChannel errorChannel = new QueueChannel(); + gateway.handleMessage(MessageBuilder.withPayload("Test1") + .setErrorChannel(errorChannel) + .build()); + Message reply = errorChannel.receive(10000); + assertThat(reply).isInstanceOf(ErrorMessage.class); + assertThat(reply.getPayload()).isInstanceOf(MessageTimeoutException.class); + doneLatch.countDown(); + gateway.stop(); + } + finally { + done.set(true); + if (ccf != null) { + ccf.stop(); + } + if (serverSocket.get() != null) { + serverSocket.get().close(); + } + sched.shutdown(); + } + } + } diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index b2d1a7758f..d293458d52 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -842,6 +842,14 @@ Starting with version 5.2, the outbound gateway can be configured with the prope 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. +Normally, the calling thread will block in the gateway, waiting for the reply (or a timeout). +Starting with version 5.3, you can set the `async` property on the gateway and the sending thread is released to do other work. +The reply (or error) will be sent on the receiving thread. +This only applies when using the `TcpNetClientConnectionFactory`, it is ignored when using NIO because there is a race condition whereby a socket error that occurs after the reply is received can be passed to the gateway before the reply. + +IMPORTANT: When using a shared connection (`singleUse=false`), a new request, while another is in process, will be blocked until the current reply is received. +Consider using the `CachingClientConnectionFactory` if you wish to support concurrent requests on a pool of long-lived connections. + [[ip-correlation]] === TCP Message Correlation @@ -1982,6 +1990,10 @@ Mutually exclusive with `remote-timeout`. | | The time in milliseconds for which the gateway waits when sending the reply to the reply-channel. Only applies if the reply-channel might block (such as a bounded QueueChannel that is currently full). + +| `async` +| +| Release the sending thread after the send; the reply (or error) will be sent on the receiving thread. |=== .IP Message Headers diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 96b8267dc4..0403e500fa 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -111,6 +111,9 @@ See <<./ws.adoc#ws,Web Services Support>> for more information. The `FailoverClientConnectionFactory` no longer fails back, by default, until the current connection fails. See <<./ip.adoc#failover-cf,TCP Failover Client Connection Factory>> for more information. +The `TcpOutboundGateway` now supports asynchronous request/reply. +See <<./ip.adoc#tcp-gateways,TCP Gateways>> for more information. + [[x5.3-rsocket]] === RSocket Changes