GH-2736: Async Mode for TcpOutboundGateway

Resolves #2736

Support asynchronous request/reply.

* - Add `async` to the schema
- Fix tests1

* - Capture `isAsync` in a variable
- Fix typo
- Convert test to JUnit5

Author: garyrussell

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 @@ protected BeanDefinitionBuilder parseHandler(Element element, ParserContext pars
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.REPLY_TIMEOUT, "sendTimeout");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "close-stream-after-send");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async");
 		return builder;
 	}
 

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,14 +39,17 @@
 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;
 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 @@ protected void doInit() {
 		}
 		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 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 		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 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 			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 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 			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 boolean onMessage(Message<?> message) {
 				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 @@ private final class AsyncReply {
 
 		private final long remoteTimeout;
 
+		private final TcpConnection connection;
+
+		private final boolean haveSemaphore;
+
+		private final SettableListenableFuture<Message<?>> 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 Message<?> getReply() {
 			return this.reply;
 		}
 
+		SettableListenableFuture<Message<?>> getFuture() {
+			return this.future;
+		}
+
 		private void doThrowErrorMessagePayload() {
 			if (this.reply.getPayload() instanceof MessagingException) {
 				throw (MessagingException) this.reply.getPayload();

spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd

@@ -464,7 +464,7 @@
 			<xsd:attribute name="remote-timeout-expression" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						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 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="async" default="false">
+				<xsd:annotation>
+					<xsd:documentation>
+						Set to true for async request/reply - see reference manual.
+					</xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:union memberTypes="xsd:boolean xsd:string"/>
+				</xsd:simpleType>
+			</xsd:attribute>
 			<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
 		</xsd:complexType>
 	</xsd:element>

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"
 		/>
 
 	<int:channel id="tcpAdviceGateChannel">

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 void testOutGateway() {
 		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