diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java index 0a8cc5df2f..3799efa292 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java +++ b/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 class GatewayEndpointSpec extends ConsumerEndpointSpec> exchange(Message request); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java index 1e28db9044..6e87f67dc9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java +++ b/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 @@ import org.springframework.messaging.MessageChannel; */ public class GatewayMessageHandler extends AbstractReplyProducingMessageHandler implements ManageableLifecycle { - private final GatewayProxyFactoryBean gatewayProxyFactoryBean; + private final Lock lock = new ReentrantLock(); - private volatile RequestReplyExchanger exchanger; + private volatile GatewayProxyFactoryBean gatewayProxyFactoryBean; + + 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 @@ public class GatewayMessageHandler extends AbstractReplyProducingMessageHandler 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); + } - if (beanFactory instanceof ConfigurableListableBeanFactory) { - ((ConfigurableListableBeanFactory) beanFactory).initializeBean(this.gatewayProxyFactoryBean, - getComponentName() + "#gpfb"); + 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 (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 @@ public class GatewayMessageHandler extends AbstractReplyProducingMessageHandler @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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/RequestReplyExchanger.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/RequestReplyExchanger.java index 4e62e32add..6ff9b3edbf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/RequestReplyExchanger.java +++ b/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 @@ import org.springframework.messaging.MessagingException; * @author Artem Bilan * * @since 2.0 + * + * @see AsyncRequestReplyExchanger */ @FunctionalInterface public interface RequestReplyExchanger { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java index 04a99f3b94..ece56118ad 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java +++ b/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 class FlowServiceTests { 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 class FlowServiceTests { @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 diff --git a/src/reference/antora/modules/ROOT/pages/dsl/java-gateway.adoc b/src/reference/antora/modules/ROOT/pages/dsl/java-gateway.adoc index 5a0faa3a27..c5a5081801 100644 --- a/src/reference/antora/modules/ROOT/pages/dsl/java-gateway.adoc +++ b/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>`, the whole request-reply is executed in asynchronous manner. diff --git a/src/reference/antora/modules/ROOT/pages/gateway.adoc b/src/reference/antora/modules/ROOT/pages/gateway.adoc index 367f0abf48..0ae3956215 100644 --- a/src/reference/antora/modules/ROOT/pages/gateway.adoc +++ b/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>`, 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` diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index a3cf9d71b7..c4da105c19 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/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`