GH-10018: Implement async for Java DSL gateway()

Fixes: #10018

The `GatewayEndpointSpec` configuration of the `gateway()` operator support already an `async(true)` option.
However, it is silently ignored internally since no real async contract provided for the gateway proxy.

* Introduce the `AsyncRequestReplyExchanger` interface to use instead of `RequestReplyExchanger`,
when  `gateway()` operator is opted-in for the `async(true)`
* Use this new `AsyncRequestReplyExchanger` in the `GatewayMessageHandler` when `async(true)`
* Also, expose a `GatewayEndpointSpec.asyncExecutor(Executor)` option to support async behavior similar to the `@MessagingGateway`

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -16,7 +16,10 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.concurrent.Executor;
+
 import org.springframework.integration.gateway.GatewayMessageHandler;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 
 /**
@@ -110,4 +113,17 @@ public GatewayEndpointSpec errorOnTimeout(boolean errorOnTimeout) {
 		return this;
 	}
 
+	/**
+	 * Set an {@link Executor} for async request-reply scenarios.
+	 * @param executor the executor to use.
+	 * @return the spec.
+	 * @since 6.5
+	 * @see org.springframework.integration.gateway.GatewayProxyFactoryBean#setAsyncExecutor(Executor)
+	 */
+
+	public GatewayEndpointSpec asyncExecutor(@Nullable Executor executor) {
+		this.handler.setAsyncExecutor(executor);
+		return this;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/gateway/AsyncRequestReplyExchanger.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2025 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.gateway;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.springframework.messaging.Message;
+
+/**
+ * Messaging gateway contract for async request/reply Message exchange.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.5
+ *
+ * @see RequestReplyExchanger
+ */
+@FunctionalInterface
+public interface AsyncRequestReplyExchanger {
+
+	CompletableFuture<Message<?>> exchange(Message<?> request);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -16,14 +16,17 @@
 
 package org.springframework.integration.gateway;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.beans.factory.BeanCreationException;
-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.support.management.ManageableLifecycle;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 
@@ -37,52 +40,79 @@
  */
 public class GatewayMessageHandler extends AbstractReplyProducingMessageHandler implements ManageableLifecycle {
 
-	private final GatewayProxyFactoryBean<?> gatewayProxyFactoryBean;
+	private final Lock lock = new ReentrantLock();
+
+	private volatile GatewayProxyFactoryBean<?> gatewayProxyFactoryBean;
 
-	private volatile RequestReplyExchanger exchanger;
+	private volatile Object exchanger;
 
 	private volatile boolean running;
 
-	private final Lock lock = new ReentrantLock();
+	private MessageChannel requestChannel;
 
-	public GatewayMessageHandler() {
-		this.gatewayProxyFactoryBean = new GatewayProxyFactoryBean<>();
-	}
+	private String requestChannelName;
+
+	private MessageChannel replyChannel;
+
+	private String replyChannelName;
+
+	private MessageChannel errorChannel;
+
+	private String errorChannelName;
+
+	private Long requestTimeout;
+
+	private Long replyTimeout;
+
+	private boolean errorOnTimeout;
+
+	private Executor executor;
 
 	public void setRequestChannel(MessageChannel requestChannel) {
-		this.gatewayProxyFactoryBean.setDefaultRequestChannel(requestChannel);
+		this.requestChannel = requestChannel;
 	}
 
 	public void setRequestChannelName(String requestChannel) {
-		this.gatewayProxyFactoryBean.setDefaultRequestChannelName(requestChannel);
+		this.requestChannelName = requestChannel;
 	}
 
 	public void setReplyChannel(MessageChannel replyChannel) {
-		this.gatewayProxyFactoryBean.setDefaultReplyChannel(replyChannel);
+		this.replyChannel = replyChannel;
 	}
 
 	public void setReplyChannelName(String replyChannel) {
-		this.gatewayProxyFactoryBean.setDefaultReplyChannelName(replyChannel);
+		this.replyChannelName = replyChannel;
 	}
 
 	public void setErrorChannel(MessageChannel errorChannel) {
-		this.gatewayProxyFactoryBean.setErrorChannel(errorChannel);
+		this.errorChannel = errorChannel;
 	}
 
 	public void setErrorChannelName(String errorChannel) {
-		this.gatewayProxyFactoryBean.setErrorChannelName(errorChannel);
+		this.errorChannelName = errorChannel;
 	}
 
 	public void setRequestTimeout(Long requestTimeout) {
-		this.gatewayProxyFactoryBean.setDefaultRequestTimeout(requestTimeout);
+		this.requestTimeout = requestTimeout;
 	}
 
 	public void setReplyTimeout(Long replyTimeout) {
-		this.gatewayProxyFactoryBean.setDefaultReplyTimeout(replyTimeout);
+		this.replyTimeout = replyTimeout;
 	}
 
 	public void setErrorOnTimeout(boolean errorOnTimeout) {
-		this.gatewayProxyFactoryBean.setErrorOnTimeout(errorOnTimeout);
+		this.errorOnTimeout = errorOnTimeout;
+	}
+
+	/**
+	 * Set the executor for use when the gateway method returns
+	 * {@link Future} or {@link CompletableFuture}.
+	 * Set it to null to disable the async processing, and any
+	 * {@link Future} return types must be returned by the downstream flow.
+	 * @param executor The executor.
+	 */
+	public void setAsyncExecutor(@Nullable Executor executor) {
+		this.executor = executor;
 	}
 
 	@Override
@@ -98,18 +128,38 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 				this.lock.unlock();
 			}
 		}
-		return this.exchanger.exchange(requestMessage);
+		return isAsync()
+				? ((AsyncRequestReplyExchanger) this.exchanger).exchange(requestMessage)
+				: ((RequestReplyExchanger) this.exchanger).exchange(requestMessage);
 	}
 
 	private void initialize() {
-		BeanFactory beanFactory = getBeanFactory();
+		if (isAsync()) {
+			this.gatewayProxyFactoryBean = new GatewayProxyFactoryBean<>(AsyncRequestReplyExchanger.class);
+		}
+		else {
+			this.gatewayProxyFactoryBean = new GatewayProxyFactoryBean<>(RequestReplyExchanger.class);
+		}
+
+		this.gatewayProxyFactoryBean.setDefaultRequestChannel(this.requestChannel);
+		this.gatewayProxyFactoryBean.setDefaultRequestChannelName(this.requestChannelName);
+		this.gatewayProxyFactoryBean.setDefaultReplyChannel(this.replyChannel);
+		this.gatewayProxyFactoryBean.setDefaultReplyChannelName(this.replyChannelName);
+		this.gatewayProxyFactoryBean.setErrorChannel(this.errorChannel);
+		this.gatewayProxyFactoryBean.setErrorChannelName(this.errorChannelName);
+		this.gatewayProxyFactoryBean.setAsyncExecutor(this.executor);
+		if (this.requestTimeout != null) {
+			this.gatewayProxyFactoryBean.setDefaultRequestTimeout(this.requestTimeout);
+		}
+		if (this.replyTimeout != null) {
+			this.gatewayProxyFactoryBean.setDefaultReplyTimeout(this.replyTimeout);
+		}
 
-		if (beanFactory instanceof ConfigurableListableBeanFactory) {
-			((ConfigurableListableBeanFactory) beanFactory).initializeBean(this.gatewayProxyFactoryBean,
-					getComponentName() + "#gpfb");
+		if (getBeanFactory() instanceof ConfigurableListableBeanFactory configurableListableBeanFactory) {
+			configurableListableBeanFactory.initializeBean(this.gatewayProxyFactoryBean, getComponentName() + "#gpfb");
 		}
 		try {
-			this.exchanger = (RequestReplyExchanger) this.gatewayProxyFactoryBean.getObject();
+			this.exchanger = this.gatewayProxyFactoryBean.getObject();
 		}
 		catch (Exception e) {
 			throw new BeanCreationException("Can't instantiate the GatewayProxyFactoryBean: " + this, e);
@@ -123,6 +173,17 @@ private void initialize() {
 
 	@Override
 	public void start() {
+		if (this.exchanger == null) {
+			this.lock.lock();
+			try {
+				if (this.exchanger == null) {
+					initialize();
+				}
+			}
+			finally {
+				this.lock.unlock();
+			}
+		}
 		this.gatewayProxyFactoryBean.start();
 		this.running = true;
 	}

spring-integration-core/src/main/java/org/springframework/integration/gateway/RequestReplyExchanger.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2025 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.
@@ -28,6 +28,8 @@
  * @author Artem Bilan
  *
  * @since 2.0
+ *
+ * @see AsyncRequestReplyExchanger
  */
 @FunctionalInterface
 public interface RequestReplyExchanger {

spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2024 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -118,6 +118,7 @@ public void testGatewayExplicitReplyChannel() {
 		Message<?> message = replyChannel.receive(10000);
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload()).isEqualTo("FOO");
+		assertThat(message.getHeaders().get("currentThread", String.class)).startsWith("SimpleAsyncTaskExecutor-");
 	}
 
 	@Autowired
@@ -155,8 +156,9 @@ public static class ContextConfiguration {
 
 		@Bean
 		public IntegrationFlow testGateway() {
-			return f -> f.gateway("processChannel", g -> g.replyChannel("replyChannel"))
-					.log();
+			return f -> f.gateway("processChannel", g -> g.replyChannel("replyChannel").async(true))
+					.enrichHeaders(headers ->
+							headers.headerExpression("currentThread", "T (Thread).currentThread().name"));
 		}
 
 		@Bean

src/reference/antora/modules/ROOT/pages/dsl/java-gateway.adoc

@@ -34,3 +34,6 @@ private static IntegrationFlow subFlow() {
 IMPORTANT: If the downstream flow does not always return a reply, you should set the `requestTimeout` to 0 to prevent hanging the calling thread indefinitely.
 In that case, the flow will end at that point and the thread released for further work.
 
+Starting with version 6.5, this `gateway()` operator fully supports an `async(true)` behaviour.
+Internally, an `AsyncRequestReplyExchanger` service interface is provided for the `GatewayProxyFactoryBean`.
+And since `AsyncRequestReplyExchanger` contract is a `CompletableFuture<Message<?>>`, the whole request-reply is executed in asynchronous manner.

src/reference/antora/modules/ROOT/pages/gateway.adoc

@@ -560,6 +560,12 @@ int finalResult =  result.get(1000, TimeUnit.SECONDS);
 
 For a more detailed example, see the https://github.com/spring-projects/spring-integration-samples/tree/main/intermediate/async-gateway[async-gateway] sample in the Spring Integration samples.
 
+Also, starting with version 6.5, the Java DSL `gateway()` operator fully supports an `async(true)` behaviour.
+Internally, an `AsyncRequestReplyExchanger` service interface is provided for the `GatewayProxyFactoryBean`.
+And since `AsyncRequestReplyExchanger` contract is a `CompletableFuture<Message<?>>`, the whole request-reply is executed in asynchronous manner.
+This behavior is useful, for example, in case of splitter-aggregator scenario when another flow has to be called for each item.
+However, the order is not important - only their group gathering on the aggregator after all processing.
+
 [[gateway-asynctaskexecutor]]
 === `AsyncTaskExecutor`
 

src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -31,6 +31,9 @@ The `AbstractMessageChannel` beans now throw a special `MessageDispatchingExcept
 In general, it is a design error to try to produce a message from `afterPropertiesSet()`, `@PostConstruct` or bean definition methods.
 The `SmartLifecycle.start()` is preferred way for this kind of logic, or better to do that via inbound channel adapters.
 
+The Java DSL `gateway()` operator now fully supports an `async(true)` behavior.
+See xref:gateway.adoc[] for more information.
+
 [[x6.5-lock-request-handler-advice]]
 == The `LockRequestHandlerAdvice`