Add ChannelBuilderCustomizer for custom gRPC Channel handling

Author: alesj

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

@@ -217,6 +217,50 @@ Don't forget to replace it with the name you used in the `@GrpcClient` annotatio
 
 IMPORTANT: When you enable `quarkus.grpc.clients."client-name".xds.enabled`, it's the xDS that should handle most of the configuration above.
 
+=== Custom Channel building
+
+When Quarkus builds a gRPC Channel instance (the way gRPC clients communicate with gRPC services on a lower network level), users can apply their own Channel(Builder) customizers. The customizers are applied by `priority`, the higher the number the later customizer is applied. The customizers are applied before Quarkus applies user's client configuration; e.g. ideal for some initial defaults per all clients.
+
+There are two `customize` methods, the first one uses gRPC's `ManagedChannelBuilder` as a parameter - to be used with Quarkus' legacy gRPC support, where the other uses `GrpcClientOptions` - to be used with the new Vert.x gRPC support. User should implement the right `customize` method per gRPC support type usage, or both if the customizer is gRPC type neutral.
+
+[source, java]
+----
+public interface ChannelBuilderCustomizer<T extends ManagedChannelBuilder<T>> {
+
+    /**
+     * Customize a ManagedChannelBuilder instance.
+     *
+     * @param name gRPC client name
+     * @param config client's configuration
+     * @param builder Channel builder instance
+     * @return map of config properties to be used as default service config against the builder
+     */
+    default Map<String, Object> customize(String name, GrpcClientConfiguration config, T builder) {
+        return Map.of();
+    }
+
+    /**
+     * Customize a GrpcClientOptions instance.
+     *
+     * @param name gRPC client name
+     * @param config client's configuration
+     * @param options GrpcClientOptions instance
+     */
+    default void customize(String name, GrpcClientConfiguration config, GrpcClientOptions options) {
+    }
+
+    /**
+     * Priority by which the customizers are applied.
+     * Higher priority is applied later.
+     *
+     * @return the priority
+     */
+    default int priority() {
+        return 0;
+    }
+}
+----
+
 === Enabling TLS
 
 To enable TLS, use the following configuration.

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

@@ -437,6 +437,11 @@ UnremovableBeanBuildItem unremovableClientInterceptors() {
         return UnremovableBeanBuildItem.beanTypes(GrpcDotNames.CLIENT_INTERCEPTOR);
     }
 
+    @BuildStep
+    UnremovableBeanBuildItem unremovableChannelBuilderCustomizers() {
+        return UnremovableBeanBuildItem.beanTypes(GrpcDotNames.CHANNEL_BUILDER_CUSTOMIZER);
+    }
+
     Set<String> getRegisteredInterceptors(InjectionPointInfo injectionPoint) {
         Set<AnnotationInstance> qualifiers = injectionPoint.getRequiredQualifiers();
         if (qualifiers.size() <= 1) {

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

@@ -24,6 +24,7 @@
 import io.quarkus.grpc.RegisterClientInterceptor;
 import io.quarkus.grpc.RegisterInterceptor;
 import io.quarkus.grpc.RegisterInterceptors;
+import io.quarkus.grpc.api.ChannelBuilderCustomizer;
 import io.quarkus.grpc.runtime.supports.Channels;
 import io.quarkus.grpc.runtime.supports.GrpcClientConfigProvider;
 import io.smallrye.common.annotation.Blocking;
@@ -58,6 +59,8 @@ public class GrpcDotNames {
             .createSimple(RegisterClientInterceptor.List.class.getName());
     public static final DotName CLIENT_INTERCEPTOR = DotName.createSimple(ClientInterceptor.class.getName());
 
+    public static final DotName CHANNEL_BUILDER_CUSTOMIZER = DotName.createSimple(ChannelBuilderCustomizer.class.getName());
+
     static final MethodDescriptor CREATE_CHANNEL_METHOD = MethodDescriptor.ofMethod(Channels.class, "createChannel",
             Channel.class, String.class, Set.class);
     static final MethodDescriptor RETRIEVE_CHANNEL_METHOD = MethodDescriptor.ofMethod(Channels.class, "retrieveChannel",

extensions/grpc/runtime/src/main/java/io/quarkus/grpc/api/ChannelBuilderCustomizer.java

@@ -0,0 +1,48 @@
+package io.quarkus.grpc.api;
+
+import java.util.Map;
+
+import io.grpc.ManagedChannelBuilder;
+import io.quarkus.grpc.runtime.config.GrpcClientConfiguration;
+import io.vertx.grpc.client.GrpcClientOptions;
+
+/**
+ * Allow for customization of Channel building.
+ * Implement the customize method, depending on which Channel implementation you're going to use,
+ * e.g. Vert.x or Netty.
+ * This is an experimental API, subject to change.
+ */
+public interface ChannelBuilderCustomizer<T extends ManagedChannelBuilder<T>> {
+
+    /**
+     * Customize a ManagedChannelBuilder instance.
+     *
+     * @param name gRPC client name
+     * @param config client's configuration
+     * @param builder Channel builder instance
+     * @return map of config properties to be used as default service config against the builder
+     */
+    default Map<String, Object> customize(String name, GrpcClientConfiguration config, T builder) {
+        return Map.of();
+    }
+
+    /**
+     * Customize a GrpcClientOptions instance.
+     *
+     * @param name gRPC client name
+     * @param config client's configuration
+     * @param options GrpcClientOptions instance
+     */
+    default void customize(String name, GrpcClientConfiguration config, GrpcClientOptions options) {
+    }
+
+    /**
+     * Priority by which the customizers are applied.
+     * Higher priority is applied later.
+     *
+     * @return the priority
+     */
+    default int priority() {
+        return 0;
+    }
+}

extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/supports/Channels.java

@@ -15,6 +15,7 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -27,6 +28,8 @@
 import java.util.concurrent.TimeoutException;
 
 import jakarta.enterprise.context.spi.CreationalContext;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.util.TypeLiteral;
 
 import org.eclipse.microprofile.context.ManagedExecutor;
 import org.jboss.logging.Logger;
@@ -50,6 +53,7 @@
 import io.quarkus.arc.InstanceHandle;
 import io.quarkus.grpc.GrpcClient;
 import io.quarkus.grpc.RegisterClientInterceptor;
+import io.quarkus.grpc.api.ChannelBuilderCustomizer;
 import io.quarkus.grpc.runtime.ClientInterceptorStorage;
 import io.quarkus.grpc.runtime.GrpcClientInterceptorContainer;
 import io.quarkus.grpc.runtime.config.GrpcClientConfiguration;
@@ -145,6 +149,13 @@ public static Channel createChannel(String name, Set<String> perClientIntercepto
             }
         }
 
+        List<ChannelBuilderCustomizer<?>> channelBuilderCustomizers = container
+                .select(new TypeLiteral<ChannelBuilderCustomizer<?>>() {
+                }, Any.Literal.INSTANCE)
+                .stream()
+                .sorted(Comparator.<ChannelBuilderCustomizer<?>, Integer> comparing(ChannelBuilderCustomizer::priority))
+                .toList();
+
         boolean plainText = config.ssl().trustStore().isEmpty();
         Optional<Boolean> usePlainText = config.plainText();
         if (usePlainText.isPresent()) {
@@ -187,8 +198,17 @@ public static Channel createChannel(String name, Set<String> perClientIntercepto
             if (provider != null) {
                 builder = provider.createChannelBuilder(config, target);
             } else {
-                builder = NettyChannelBuilder
-                        .forTarget(target)
+                builder = NettyChannelBuilder.forTarget(target);
+            }
+
+            for (ChannelBuilderCustomizer customizer : channelBuilderCustomizers) {
+                Map<String, Object> map = customizer.customize(name, config, builder);
+                builder.defaultServiceConfig(map);
+            }
+
+            if (builder instanceof NettyChannelBuilder) {
+                NettyChannelBuilder ncBuilder = (NettyChannelBuilder) builder;
+                builder = ncBuilder
                         // clients are intercepted using the IOThreadClientInterceptor interceptor which will decide on which
                         // thread the messages should be processed.
                         .directExecutor() // will use I/O thread - must not be blocked.
@@ -201,6 +221,10 @@ public static Channel createChannel(String name, Set<String> perClientIntercepto
                         .maxInboundMetadataSize(config.maxInboundMetadataSize().orElse(DEFAULT_MAX_HEADER_LIST_SIZE))
                         .maxInboundMessageSize(config.maxInboundMessageSize().orElse(DEFAULT_MAX_MESSAGE_SIZE))
                         .negotiationType(NegotiationType.valueOf(config.negotiationType().toUpperCase()));
+
+                if (context != null) {
+                    ncBuilder.sslContext(context);
+                }
             }
 
             if (config.retry()) {
@@ -242,10 +266,6 @@ public static Channel createChannel(String name, Set<String> perClientIntercepto
             if (plainText && provider == null) {
                 builder.usePlaintext();
             }
-            if (context != null && (builder instanceof NettyChannelBuilder)) {
-                NettyChannelBuilder ncBuilder = (NettyChannelBuilder) builder;
-                ncBuilder.sslContext(context);
-            }
 
             interceptorContainer.getSortedPerServiceInterceptors(perClientInterceptors).forEach(builder::intercept);
             interceptorContainer.getSortedGlobalInterceptors().forEach(builder::intercept);
@@ -259,6 +279,15 @@ public static Channel createChannel(String name, Set<String> perClientIntercepto
             HttpClientOptions options = new HttpClientOptions();
             options.setHttp2ClearTextUpgrade(false); // this fixes i30379
 
+            // Start with almost empty options and default max msg size ...
+            GrpcClientOptions clientOptions = new GrpcClientOptions()
+                    .setTransportOptions(options)
+                    .setMaxMessageSize(config.maxInboundMessageSize().orElse(DEFAULT_MAX_MESSAGE_SIZE));
+
+            for (ChannelBuilderCustomizer customizer : channelBuilderCustomizers) {
+                customizer.customize(name, config, clientOptions);
+            }
+
             if (!plainText) {
                 TlsConfigurationRegistry registry = Arc.container().select(TlsConfigurationRegistry.class).get();
 
@@ -330,9 +359,10 @@ public static Channel createChannel(String name, Set<String> perClientIntercepto
             options.setMetricsName("grpc|" + name);
 
             Vertx vertx = container.instance(Vertx.class).get();
-            io.vertx.grpc.client.GrpcClient client = io.vertx.grpc.client.GrpcClient.client(vertx,
-                    new GrpcClientOptions().setTransportOptions(options)
-                            .setMaxMessageSize(config.maxInboundMessageSize().orElse(DEFAULT_MAX_MESSAGE_SIZE)));
+            io.vertx.grpc.client.GrpcClient client = io.vertx.grpc.client.GrpcClient.client(
+                    vertx,
+                    clientOptions);
+
             Channel channel;
             if (stork) {
                 ManagedExecutor executor = container.instance(ManagedExecutor.class).get();

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

@@ -32,6 +32,7 @@ public Response helloBlocking(@PathParam("name") String name) {
 
         return Response.ok(helloReply.getMessage())
                 .header("interceptors", String.join(",", invoked))
+                .header("used", MyCBC.USED.get() + "")
                 .build();
     }
 

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

@@ -0,0 +1,27 @@
+package io.quarkus.grpc.examples.interceptors;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import io.grpc.netty.NettyChannelBuilder;
+import io.quarkus.grpc.api.ChannelBuilderCustomizer;
+import io.quarkus.grpc.runtime.config.GrpcClientConfiguration;
+import io.vertx.grpc.client.GrpcClientOptions;
+
+@ApplicationScoped
+public class MyCBC implements ChannelBuilderCustomizer<NettyChannelBuilder> {
+    public static final AtomicBoolean USED = new AtomicBoolean(false);
+
+    @Override
+    public Map<String, Object> customize(String name, GrpcClientConfiguration config, NettyChannelBuilder builder) {
+        USED.set(true);
+        return Map.of();
+    }
+
+    @Override
+    public void customize(String name, GrpcClientConfiguration config, GrpcClientOptions options) {
+        USED.set(true);
+    }
+}

integration-tests/grpc-interceptors/src/test/java/io/quarkus/grpc/example/interceptors/HelloWorldEndpointTestBase.java

@@ -26,6 +26,9 @@ public void testHelloWorldServiceUsingBlockingStub() {
                 "io.quarkus.grpc.examples.interceptors.ServerInterceptors$MethodTarget");
 
         ensureThatMetricsAreProduced();
+
+        String used = response.getHeader("used");
+        assertThat(Boolean.parseBoolean(used)).isTrue();
     }
 
     @Test