Merge pull request #47094 from alesj/grpc_reqctx1

Allow for RoutingContext to be injected into gRPC service

Author: cescoffier

docs/src/main/asciidoc/grpc-service-implementation.adoc

@@ -331,6 +331,21 @@ public class MyInterceptor implements ServerInterceptor, Prioritized {
 Interceptors with the highest priority are called first.
 The default priority, used if the interceptor does not implement the `Prioritized` interface, is `0`.
 
+There is also a support to inject Vert.x `RoutingContext` instance into your gRPC service, if / when needed.
+Quarkus doesn't do that by default, you will need to add `RoutingContextGrpcInterceptor` to your gRPC service.
+
+[source, java]
+----
+@GrpcService
+@RegisterInterceptor(RoutingContextGrpcInterceptor.class)
+public class HelloWorldService extends GreeterGrpc.GreeterImplBase {
+
+    @Inject
+    RoutingContext context;
+
+    // ...
+}
+----
 
 == Testing your services
 

extensions/grpc/deployment/src/main/java/io/quarkus/grpc/deployment/GrpcServerProcessor.java

@@ -86,6 +86,7 @@
 import io.quarkus.grpc.runtime.health.GrpcHealthStorage;
 import io.quarkus.grpc.runtime.supports.context.GrpcDuplicatedContextGrpcInterceptor;
 import io.quarkus.grpc.runtime.supports.context.GrpcRequestContextGrpcInterceptor;
+import io.quarkus.grpc.runtime.supports.context.RoutingContextGrpcInterceptor;
 import io.quarkus.grpc.runtime.supports.exc.DefaultExceptionHandlerProvider;
 import io.quarkus.grpc.runtime.supports.exc.ExceptionInterceptor;
 import io.quarkus.kubernetes.spi.KubernetesPortBuildItem;
@@ -565,6 +566,7 @@ void registerBeans(BuildProducer<AdditionalBeanBuildItem> beans,
             // Global interceptors are invoked before any of the per-service interceptors
             beans.produce(AdditionalBeanBuildItem.unremovableOf(GrpcRequestContextGrpcInterceptor.class));
             beans.produce(AdditionalBeanBuildItem.unremovableOf(GrpcDuplicatedContextGrpcInterceptor.class));
+            beans.produce(AdditionalBeanBuildItem.unremovableOf(RoutingContextGrpcInterceptor.class));
             features.produce(new FeatureBuildItem(GRPC_SERVER));
 
             if (capabilities.isPresent(Capability.SECURITY)) {

extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java

@@ -269,20 +269,30 @@ public void handle(Void unused) {
                         capturedVertxContext.runOnContext(new Handler<Void>() {
                             @Override
                             public void handle(Void unused) {
-                                server.handle(ctx.request());
+                                routingContextAware(server, ctx);
                             }
                         });
                         return;
                     }
                 }
-                server.handle(ctx.request());
+                routingContextAware(server, ctx);
             }
         });
         shutdown.addShutdownTask(route::remove); // remove this route at shutdown, this should reset it
 
         initHealthStorage();
     }
 
+    private static void routingContextAware(GrpcServer server, RoutingContext context) {
+        Context currentContext = Vertx.currentContext();
+        currentContext.putLocal(RoutingContext.class.getName(), context);
+        try {
+            server.handle(context.request());
+        } finally {
+            currentContext.removeLocal(RoutingContext.class.getName());
+        }
+    }
+
     // TODO -- handle Avro, plain text ... when supported / needed
     private static boolean isGrpc(RoutingContext rc) {
         HttpServerRequest request = rc.request();

extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/Interceptors.java

@@ -16,6 +16,7 @@ public final class Interceptors {
 
     public static final int DUPLICATE_CONTEXT = Integer.MAX_VALUE - 5;
     public static final int REQUEST_CONTEXT = Integer.MAX_VALUE - 50;
+    public static final int ROUTING_CONTEXT = Integer.MAX_VALUE - 55;
     public static final int BLOCKING_HANDLER = Integer.MAX_VALUE - 60;
     public static final int EXCEPTION_HANDLER = Integer.MAX_VALUE - 70;
 

extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/supports/context/RoutingContextGrpcInterceptor.java

@@ -0,0 +1,84 @@
+package io.quarkus.grpc.runtime.supports.context;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.spi.Prioritized;
+
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.ArcContainer;
+import io.quarkus.arc.ManagedContext;
+import io.quarkus.grpc.runtime.Interceptors;
+import io.quarkus.vertx.http.runtime.CurrentVertxRequest;
+import io.vertx.core.Context;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.RoutingContext;
+
+@ApplicationScoped
+public class RoutingContextGrpcInterceptor implements ServerInterceptor, Prioritized {
+
+    @Override
+    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+            ServerCall<ReqT, RespT> call,
+            Metadata headers,
+            ServerCallHandler<ReqT, RespT> next) {
+
+        Context currentContext = Vertx.currentContext();
+        RoutingContext routingContext = currentContext.getLocal(RoutingContext.class.getName());
+
+        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
+                next.startCall(call, headers)) {
+
+            private void invoke(Runnable runnable) {
+                ArcContainer container = Arc.container();
+                ManagedContext reqContext = container.requestContext();
+                CurrentVertxRequest cvr = null;
+                if (reqContext.isActive()) {
+                    cvr = container.select(CurrentVertxRequest.class).get();
+                    cvr.setCurrent(routingContext);
+                }
+                try {
+                    runnable.run();
+                } finally {
+                    // can be non-active, after onComplete + handleClose -> onHalfClose
+                    if (cvr != null && reqContext.isActive()) {
+                        cvr.setCurrent(null);
+                    }
+                }
+            }
+
+            @Override
+            public void onMessage(ReqT message) {
+                invoke(() -> super.onMessage(message));
+            }
+
+            @Override
+            public void onHalfClose() {
+                invoke(super::onHalfClose);
+            }
+
+            @Override
+            public void onCancel() {
+                invoke(super::onCancel);
+            }
+
+            @Override
+            public void onComplete() {
+                invoke(super::onComplete);
+            }
+
+            @Override
+            public void onReady() {
+                invoke(super::onReady);
+            }
+        };
+    }
+
+    @Override
+    public int getPriority() {
+        return Interceptors.ROUTING_CONTEXT;
+    }
+}

integration-tests/grpc-interceptors/src/main/java/io/quarkus/grpc/examples/interceptors/HelloWorldService.java

@@ -1,15 +1,35 @@
 package io.quarkus.grpc.examples.interceptors;
 
+import jakarta.inject.Inject;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
 import examples.GreeterGrpc;
 import examples.HelloReply;
 import examples.HelloRequest;
 import io.grpc.stub.StreamObserver;
 import io.quarkus.grpc.GrpcService;
+import io.quarkus.grpc.RegisterInterceptor;
+import io.quarkus.grpc.runtime.supports.context.RoutingContextGrpcInterceptor;
+import io.vertx.ext.web.RoutingContext;
 
 @GrpcService
+@RegisterInterceptor(RoutingContextGrpcInterceptor.class)
 public class HelloWorldService extends GreeterGrpc.GreeterImplBase {
 
+    @Inject
+    RoutingContext context;
+
+    @ConfigProperty(name = "quarkus.profile")
+    String profile;
+
     private HelloReply getReply(HelloRequest request) {
+        // just poke for Vert.x based gRPC server-side
+        if ("vertx".equals(profile) || "o2n".equals(profile)) {
+            // just poke context, so we know it works
+            context.request().getHeader("foobar");
+        }
+
         String name = request.getName();
         if (name.equals("Fail")) {
             throw new HelloException(name);