GH-2736: TCP Async GW - Fix async check

Checking for NIO cannot be done in `onInit()` because the factory is
not yet started.

**cherry-pick to 5.3.x**

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

@@ -39,7 +39,6 @@
 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;
@@ -127,21 +126,6 @@ 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);
-			}
-		}
 	}
 
 	/**
@@ -167,6 +151,7 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 		try {
 			haveSemaphore = acquireSemaphoreIfNeeded(requestMessage);
 			connection = this.connectionFactory.getConnection();
+			checkAsync(connection, async);
 			Long remoteTimeout = getRemoteTimeout(requestMessage);
 			AsyncReply reply = new AsyncReply(remoteTimeout, connection, haveSemaphore, requestMessage, async);
 			connectionId = connection.getConnectionId();
@@ -203,6 +188,13 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 		}
 	}
 
+	private void checkAsync(TcpConnection connection, boolean async) {
+		if (async && connection instanceof TcpNioConnectionSupport) {
+			setAsync(false);
+			this.logger.warn("Async replies are not supported with NIO; see the reference manual");
+		}
+	}
+
 	private boolean acquireSemaphoreIfNeeded(Message<?> requestMessage) throws InterruptedException {
 		if (!this.isSingleUse) {
 			logger.debug("trying semaphore");