GH-2736: Add DSL support for async TCP OB Gateway

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

@@ -68,7 +68,7 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
 	 */
 	public TcpOutboundGatewaySpec remoteTimeout(long remoteTimeout) {
 		this.target.setRemoteTimeout(remoteTimeout);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -86,7 +86,7 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
 	 */
 	public <P> TcpOutboundGatewaySpec remoteTimeout(Function<Message<P>, ?> remoteTimeoutFunction) {
 		this.target.setRemoteTimeoutExpression(new FunctionExpression<>(remoteTimeoutFunction));
-		return _this();
+		return this;
 	}
 
 	/**
@@ -100,7 +100,18 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
 	 */
 	public TcpOutboundGatewaySpec closeStreamAfterSend(boolean closeStreamAfterSend) {
 		this.target.setCloseStreamAfterSend(closeStreamAfterSend);
-		return _this();
+		return this;
+	}
+
+	/**
+	 * Set to true to release the sending thread and receive the reply asynchronously.
+	 * @param async true for asynchronous request/reply.
+	 * @return the spec.
+	 * @since 5.3
+	 */
+	public TcpOutboundGatewaySpec async(boolean async) {
+		this.target.setAsync(async);
+		return this;
 	}
 
 	@Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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 @@ import org.springframework.integration.ip.udp.UdpServerListeningEvent;
 import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
 import org.springframework.integration.ip.util.TestingUtilities;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
@@ -100,6 +101,9 @@ public class IpIntegrationTests {
 	@Autowired
 	private QueueChannel udpIn;
 
+	@Autowired
+	private TcpOutboundGateway tcpOutAsync;
+
 	@Autowired
 	private Config config;
 
@@ -210,6 +214,11 @@ public class IpIntegrationTests {
 				.convertSendAndReceive("foo", String.class)).isEqualTo("reply:FOO");
 	}
 
+	@Test
+	void async() {
+		assertThat(TestUtils.getPropertyValue(this.tcpOutAsync, "async", Boolean.class)).isTrue();
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class Config {
@@ -293,6 +302,22 @@ public class IpIntegrationTests {
 					.get();
 		}
 
+		@Bean
+		public AbstractClientConnectionFactory client2() {
+			return Tcp.netClient("localhost", server1().getPort())
+					.serializer(TcpCodecs.crlf())
+					.deserializer(TcpCodecs.lengthHeader1())
+					.get();
+		}
+
+		@Bean
+		public TcpOutboundGateway tcpOutAsync() {
+			return Tcp.outboundGateway(client2())
+					.async(true)
+					.remoteTimeout(m -> 5000)
+					.get();
+		}
+
 		@Bean
 		public AtomicBoolean adviceCalled() {
 			return new AtomicBoolean();