From 7fa72282c1a353b66471f439ac33937d976b361d Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 14 Feb 2024 20:54:44 +0100 Subject: [PATCH] GH-2898 Add support for asyn send to StreamBridge --- core/spring-cloud-stream/pom.xml | 6 +++ .../cloud/stream/function/StreamBridge.java | 51 +++++++++++++++++-- .../producing-and-consuming-messages.adoc | 18 +++++++ 3 files changed, 72 insertions(+), 3 deletions(-) diff --git a/core/spring-cloud-stream/pom.xml b/core/spring-cloud-stream/pom.xml index c573abe70..3d9c6b2e6 100644 --- a/core/spring-cloud-stream/pom.xml +++ b/core/spring-cloud-stream/pom.xml @@ -65,6 +65,12 @@ spring-core-test test + + io.micrometer + context-propagation + 1.1.0 + provided + org.springframework.boot spring-boot-autoconfigure-processor diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 3f9737ecf..1d470468a 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -21,8 +21,13 @@ import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.function.Function; +import io.micrometer.context.ContextExecutorService; +import io.micrometer.context.ContextSnapshotFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -47,6 +52,8 @@ import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.ResolvableType; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.AbstractSubscribableChannel; +import org.springframework.integration.channel.ExecutorChannel; import org.springframework.integration.config.GlobalChannelInterceptorProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.lang.Nullable; @@ -54,11 +61,13 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; + /** * A class which allows user to send data to an output binding. * While in a common scenario of a typical spring-cloud-stream application user rarely @@ -96,12 +105,19 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi private boolean initialized; + private boolean async; + private final BindingService bindingService; private final Map streamBridgeFunctionCache; private final FunctionInvocationHelper functionInvocationHelper; + private ExecutorService executorService; + + private static final boolean isContextPropagationPresent = ClassUtils.isPresent( + "io.micrometer.context.ContextSnapshotFactory", StreamBridge.class.getClassLoader()); + /** * * @param functionCatalog instance of {@link FunctionCatalog} @@ -111,6 +127,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi @SuppressWarnings("serial") StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) { + this.executorService = Executors.newCachedThreadPool(); Assert.notNull(functionCatalog, "'functionCatalog' must not be null"); Assert.notNull(applicationContext, "'applicationContext' must not be null"); Assert.notNull(bindingServiceProperties, "'bindingServiceProperties' must not be null"); @@ -253,9 +270,9 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi } } else { - messageChannel = new DirectWithAttributesChannel(); - ((DirectWithAttributesChannel) messageChannel).setApplicationContext(applicationContext); - ((DirectWithAttributesChannel) messageChannel).setComponentName(destinationName); + messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel(); + ((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext); + ((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName); if (this.destinationBindingCallback != null) { Object extendedProducerProperties = this.bindingService .getExtendedProducerProperties(messageChannel, destinationName); @@ -310,8 +327,36 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi @Override public void destroy() throws Exception { + if (!this.executorService.awaitTermination(10000, TimeUnit.MILLISECONDS)) { + logger.warn("Failed to terminate executor. Terminating current tasks."); + this.executorService.shutdownNow(); + } + else { + this.executorService.shutdown(); + } + + this.executorService = null; + this.async = false; channelCache.keySet().forEach(bindingService::unbindProducers); channelCache.clear(); } + public boolean isAsync() { + return async; + } + + public void setAsync(boolean async) { + if (isContextPropagationPresent) { + this.executorService = ContextPropagationHelper.wrap(this.executorService); + } + this.executorService = ContextExecutorService + .wrap(Executors.newCachedThreadPool(), () -> ContextSnapshotFactory.builder().build().captureAll()); + this.async = async; + } + + private static final class ContextPropagationHelper { + static ExecutorService wrap(ExecutorService executorService) { + return ContextExecutorService.wrap(executorService, () -> ContextSnapshotFactory.builder().build().captureAll()); + } + } } diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc index 78271c98a..d919f0aa8 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc @@ -293,6 +293,24 @@ Also, note that `streamBridge.send(..)` method takes an `Object` for data. This will go through the same routine when sending output as if it was from any Function or Supplier providing the same level of consistency as with functions. This means the output type conversion, partitioning etc are honored as if it was from the output produced by functions. + +==== StreamBridge with async send + +`StreamBridge` uses sending mechanism provided by _Spring Integration_ framework which is at the core of the _Spring Cloud Stream_. By default this mechanism uses the sender’s thread. In other words, the send is blocking. While this is ok for many cases, there are cases when you want such send to be async. To do that use `setAsync(true)` method of the `StreamBridge` before invoking one of the send methods. + +**Observability Context propagation with asynchronous send** + +When using Observability support provided by the framework as well as supporting Spring frameworks, breaking thread boundaries will affect consistency of Observability context, thus your tracing history. To avoid that all you need is to add `context-propagation` dependency form Micrometer (see below) + +[source, xml] +---- + + io.micrometer + context-propagation + 1.1.0 + +---- + [[streambridge-and-dynamic-destinations]] ==== StreamBridge and Dynamic Destinations