Merge pull request #45827 from ozangunalp/messaging_context_propagation

Context propagation for Messaging

Author: cescoffier

docs/src/main/asciidoc/messaging.adoc

@@ -562,6 +562,7 @@ public class StreamProcessor {
 }
 ----
 
+[[execution_model]]
 == Execution Model
 
 Quarkus Messaging sits on top of the xref:quarkus-reactive-architecture.adoc#engine[reactive engine] of Quarkus and leverages link:{eclipse-vertx}[Eclipse Vert.x] to dispatch messages for processing.
@@ -634,6 +635,100 @@ Depending on the broker technology, this can be useful to increase the applicati
 while still preserving the partial order of messages received in different copies.
 This is the case, for example, for Kafka, where multiple consumers can consume different topic partitions.
 
+== Context Propagation
+
+In Quarkus Messaging, the default mechanism for propagating context between different processing stages is the
+link:https://smallrye.io/smallrye-reactive-messaging/latest/concepts/message-context[message context].
+This provides a consistent way to pass context information along with the message as it flows through different stages.
+
+=== Interaction with Mutiny and MicroProfile Context Propagation
+
+Mutiny, which is the foundation of reactive programming in Quarkus, is integrated with the MicroProfile Context Propagation.
+This integration enables automatic capturing and restoring of context across asynchronous boundaries.
+To learn more about context propagation in Quarkus and Mutiny, refer to the xref:context-propagation.adoc[Context Propagation] guide.
+
+However, Quarkus Messaging needs to coordinate multiple asynchronous boundaries.
+This is why the default context propagation can result in unexpected behavior in some cases, especially using `Emitters`.
+
+To ensure consistent behavior, Quarkus Messaging disables the propagation of any context during message dispatching, through internal channels or connectors.
+This means that Emitters won't capture the caller context, and incoming channels won't dispatch messages by activating a context (ex. the request context).
+
+For example, you might want to propagate the caller context from an incoming HTTP request to the message processing stage.
+For emitters, instead of using the regular `Emitter` or `MutinyEmitter`, you can inject the `ContextualEmitter` to make sure the message captures the caller context.
+This ensures consistent and predictable behaviour by relying on the message context handling provided by the framework.
+
+For example, let `RequestScopedBean` a request-scoped bean, `ContextualEmitter` can be used to dispatch messages locally through the internal channel `app`:
+
+[source, java]
+----
+import jakarta.inject.Inject;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.core.MediaType;
+
+import org.eclipse.microprofile.reactive.messaging.Channel;
+
+import io.quarkus.logging.Log;
+import io.quarkus.smallrye.reactivemessaging.runtime.ContextualEmitter;
+import io.smallrye.mutiny.Uni;
+import io.vertx.core.Context;
+import io.vertx.core.Vertx;
+
+@Path("/")
+public class Resource {
+
+    @Channel("app")
+    ContextualEmitter<String> emitter;
+
+    @Inject
+    RequestScopedBean requestScopedBean;
+
+    @POST
+    @Path("/send")
+    public void send(String message) {
+        requestScopedBean.setValue("Hello");
+        emitter.sendAndAwait(message);
+    }
+
+}
+----
+
+Then the request-scoped bean can be accessed in the message processing stage, regardless of the <<execution_model>>:
+
+[source, java]
+----
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+
+import io.quarkus.logging.Log;
+import io.smallrye.reactive.messaging.annotations.Blocking;
+
+
+@ApplicationScoped
+public class Processor {
+
+    @Inject
+    RequestScopedBean requestScopedBean;
+
+    @Incoming("app")
+    @Blocking
+    public void process(String message) {
+        Log.infof("Message %s from request %s", message, requestScopedBean.getValue());
+    }
+
+}
+----
+
+=== Request Context Activation
+
+In some cases, you might need to activate the request context while processing messages consumed from a broker.
+While using `@ActivateRequestContext` on the `@Incoming` method is an option, it's lifecycle does not follow that of a Quarkus Messaging message.
+For incoming channels, you can enable the request scope activation with the build time property `quarkus.messaging.request-scoped.enabled=true`.
+This will activate the request context for each message processed by the incoming channel, and close the context once the message is processed.
+
 == Health Checks
 
 Together with the SmallRye Health extension, Quarkus Messaging extensions provide health check support per channel.

extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingBuildTimeConfig.java

@@ -27,4 +27,11 @@ public interface ReactiveMessagingBuildTimeConfig {
     @WithName("auto-connector-attachment")
     @WithDefault("true")
     boolean autoConnectorAttachment();
+
+    /**
+     * Whether to enable the RequestScope context on a message context
+     */
+    @WithName("request-scoped.enabled")
+    @WithDefault("false")
+    boolean activateRequestScopeEnabled();
 }

extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java

@@ -70,13 +70,16 @@
 import io.quarkus.smallrye.reactivemessaging.deployment.items.InjectedChannelBuildItem;
 import io.quarkus.smallrye.reactivemessaging.deployment.items.InjectedEmitterBuildItem;
 import io.quarkus.smallrye.reactivemessaging.deployment.items.MediatorBuildItem;
+import io.quarkus.smallrye.reactivemessaging.runtime.ContextClearedDecorator;
+import io.quarkus.smallrye.reactivemessaging.runtime.ContextualEmitterFactory;
 import io.quarkus.smallrye.reactivemessaging.runtime.DuplicatedContextConnectorFactory;
 import io.quarkus.smallrye.reactivemessaging.runtime.DuplicatedContextConnectorFactoryInterceptor;
 import io.quarkus.smallrye.reactivemessaging.runtime.HealthCenterFilter;
 import io.quarkus.smallrye.reactivemessaging.runtime.HealthCenterInterceptor;
 import io.quarkus.smallrye.reactivemessaging.runtime.QuarkusMediatorConfiguration;
 import io.quarkus.smallrye.reactivemessaging.runtime.QuarkusWorkerPoolRegistry;
 import io.quarkus.smallrye.reactivemessaging.runtime.ReactiveMessagingConfiguration;
+import io.quarkus.smallrye.reactivemessaging.runtime.RequestScopedDecorator;
 import io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle;
 import io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingRecorder;
 import io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingRecorder.SmallRyeReactiveMessagingContext;
@@ -112,11 +115,14 @@ FeatureBuildItem feature() {
     }
 
     @BuildStep
-    AdditionalBeanBuildItem beans() {
+    void beans(BuildProducer<AdditionalBeanBuildItem> additionalBean, ReactiveMessagingBuildTimeConfig buildTimeConfig) {
         // We add the connector and channel qualifiers to make them part of the index.
-        return new AdditionalBeanBuildItem(SmallRyeReactiveMessagingLifecycle.class, Connector.class,
+        additionalBean.produce(new AdditionalBeanBuildItem(SmallRyeReactiveMessagingLifecycle.class, Connector.class,
                 Channel.class, io.smallrye.reactive.messaging.annotations.Channel.class,
-                QuarkusWorkerPoolRegistry.class);
+                QuarkusWorkerPoolRegistry.class, ContextualEmitterFactory.class, ContextClearedDecorator.class));
+        if (buildTimeConfig.activateRequestScopeEnabled()) {
+            additionalBean.produce(new AdditionalBeanBuildItem(RequestScopedDecorator.class));
+        }
     }
 
     @BuildStep

extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/ContextClearedDecorator.java

@@ -0,0 +1,39 @@
+package io.quarkus.smallrye.reactivemessaging.runtime;
+
+import java.util.List;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import org.eclipse.microprofile.context.ThreadContext;
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.reactive.messaging.PublisherDecorator;
+
+@ApplicationScoped
+public class ContextClearedDecorator implements PublisherDecorator {
+
+    private final ThreadContext tc;
+
+    public ContextClearedDecorator() {
+        tc = ThreadContext.builder()
+                .propagated(ThreadContext.NONE)
+                .cleared(ThreadContext.ALL_REMAINING)
+                .build();
+    }
+
+    @Override
+    public Multi<? extends Message<?>> decorate(Multi<? extends Message<?>> publisher, List<String> channelName,
+            boolean isConnector) {
+        if (isConnector) {
+            return publisher.emitOn(tc.currentContextExecutor());
+        }
+        return publisher;
+    }
+
+    @Override
+    public int getPriority() {
+        // Before the io.smallrye.reactive.messaging.providers.locals.ContextDecorator which has the priority 0
+        return -100;
+    }
+}

extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/ContextualEmitter.java

@@ -0,0 +1,17 @@
+package io.quarkus.smallrye.reactivemessaging.runtime;
+
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+import io.smallrye.mutiny.Uni;
+import io.smallrye.reactive.messaging.EmitterType;
+
+public interface ContextualEmitter<T> extends EmitterType {
+
+    Uni<Void> send(T payload);
+
+    void sendAndAwait(T payload);
+
+    <M extends Message<? extends T>> Uni<Void> sendMessage(M msg);
+
+    <M extends Message<? extends T>> void sendMessageAndAwait(M msg);
+}

extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/ContextualEmitterFactory.java

@@ -0,0 +1,36 @@
+package io.quarkus.smallrye.reactivemessaging.runtime;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import jakarta.enterprise.inject.Typed;
+import jakarta.enterprise.inject.spi.InjectionPoint;
+import jakarta.inject.Inject;
+
+import org.eclipse.microprofile.reactive.messaging.Channel;
+
+import io.smallrye.reactive.messaging.ChannelRegistry;
+import io.smallrye.reactive.messaging.EmitterConfiguration;
+import io.smallrye.reactive.messaging.EmitterFactory;
+import io.smallrye.reactive.messaging.annotations.EmitterFactoryFor;
+import io.smallrye.reactive.messaging.providers.extension.ChannelProducer;
+
+@EmitterFactoryFor(ContextualEmitter.class)
+@ApplicationScoped
+public class ContextualEmitterFactory implements EmitterFactory<ContextualEmitterImpl<Object>> {
+
+    @Inject
+    ChannelRegistry channelRegistry;
+
+    @Override
+    public ContextualEmitterImpl<Object> createEmitter(EmitterConfiguration emitterConfiguration, long l) {
+        return new ContextualEmitterImpl<>(emitterConfiguration, l);
+    }
+
+    @Produces
+    @Typed(ContextualEmitter.class)
+    @Channel("") // Stream name is ignored during type-safe resolution
+    <T> ContextualEmitter<T> produceEmitter(InjectionPoint injectionPoint) {
+        String channelName = ChannelProducer.getChannelName(injectionPoint);
+        return channelRegistry.getEmitter(channelName, ContextualEmitter.class);
+    }
+}

extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/ContextualEmitterImpl.java

@@ -0,0 +1,72 @@
+package io.quarkus.smallrye.reactivemessaging.runtime;
+
+import static io.smallrye.reactive.messaging.providers.i18n.ProviderExceptions.ex;
+
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+import io.smallrye.common.annotation.CheckReturnValue;
+import io.smallrye.common.vertx.VertxContext;
+import io.smallrye.mutiny.Uni;
+import io.smallrye.reactive.messaging.EmitterConfiguration;
+import io.smallrye.reactive.messaging.providers.extension.AbstractEmitter;
+import io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata;
+import io.vertx.core.Context;
+import io.vertx.core.Vertx;
+
+public class ContextualEmitterImpl<T> extends AbstractEmitter<T> implements ContextualEmitter<T> {
+
+    public ContextualEmitterImpl(EmitterConfiguration configuration, long defaultBufferSize) {
+        super(configuration, defaultBufferSize);
+    }
+
+    @Override
+    public void sendAndAwait(T payload) {
+        sendMessage(Message.of(payload)).await().indefinitely();
+    }
+
+    @Override
+    public Uni<Void> send(T payload) {
+        return sendMessage(Message.of(payload));
+    }
+
+    @Override
+    public <M extends Message<? extends T>> void sendMessageAndAwait(M msg) {
+        sendMessage(msg).await().indefinitely();
+    }
+
+    @Override
+    @CheckReturnValue
+    public <M extends Message<? extends T>> Uni<Void> sendMessage(M msg) {
+        if (msg == null) {
+            throw ex.illegalArgumentForNullValue();
+        }
+
+        // If we are running on a Vert.x context, we need to capture the context to switch back during the emission.
+        Context context = Vertx.currentContext();
+        Uni<Void> uni = Uni.createFrom().emitter(e -> {
+            try {
+                Message<? extends T> message = msg;
+                if (VertxContext.isDuplicatedContext(context)) {
+                    message = message.addMetadata(new LocalContextMetadata(context));
+                }
+                emit(message.withAck(() -> {
+                    e.complete(null);
+                    return msg.ack();
+                })
+                        .withNack(t -> {
+                            e.fail(t);
+                            return msg.nack(t);
+                        }));
+            } catch (Exception t) {
+                // Capture synchronous exception and nack the message.
+                msg.nack(t);
+                throw t;
+            }
+        });
+        if (context != null) {
+            uni = uni.emitOn(runnable -> context.runOnContext(x -> runnable.run()));
+        }
+        return uni;
+    }
+
+}