- code cleanup, bug fixing

- ref #3875

Author: jknack

jooby/src/main/java/io/jooby/GrpcExchange.java

@@ -9,12 +9,14 @@
 import java.util.Map;
 import java.util.function.Consumer;
 
+import edu.umd.cs.findbugs.annotations.Nullable;
+
 /** Server-agnostic abstraction for HTTP/2 trailing-header exchanges. */
 public interface GrpcExchange {
 
   String getRequestPath();
 
-  String getHeader(String name);
+  @Nullable String getHeader(String name);
 
   Map<String, String> getHeaders();
 

jooby/src/main/java/io/jooby/GrpcProcessor.java

@@ -12,6 +12,10 @@
 
 /** Intercepts and processes gRPC exchanges. */
 public interface GrpcProcessor {
+
+  /** Checks if the given URI path exactly matches a registered gRPC method. */
+  boolean isGrpcMethod(String path);
+
   /**
    * @return A subscriber that the server will feed ByteBuffer chunks into, or null if the exchange
    *     was rejected/unimplemented.

modules/jooby-grpc/src/main/java/io/jooby/grpc/GrpcMethodRegistry.java

@@ -5,22 +5,24 @@
  */
 package io.jooby.grpc;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.slf4j.bridge.SLF4JBridgeHandler;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 import io.grpc.BindableService;
+import io.grpc.MethodDescriptor;
 import io.grpc.Server;
 import io.grpc.inprocess.InProcessChannelBuilder;
 import io.grpc.inprocess.InProcessServerBuilder;
-import io.grpc.protobuf.services.ProtoReflectionServiceV1;
 import io.jooby.*;
+import io.jooby.internal.grpc.DefaultGrpcProcessor;
 
 public class GrpcModule implements Extension {
   private final List<BindableService> services;
-  private final GrpcMethodRegistry methodRegistry = new GrpcMethodRegistry();
-  private final String serverName = "jooby-internal-" + System.nanoTime();
+  private final Map<String, MethodDescriptor<?, ?>> registry = new HashMap<>();
   private Server grpcServer;
 
   static {
@@ -36,35 +38,47 @@ public GrpcModule(BindableService... services) {
 
   @Override
   public void install(@NonNull Jooby app) throws Exception {
-    var builder = InProcessServerBuilder.forName(serverName).directExecutor();
+    var serverName = app.getName();
+    var builder = InProcessServerBuilder.forName(serverName);
 
     // 1. Register user-provided services
-    for (BindableService service : services) {
+    for (var service : services) {
       builder.addService(service);
-      methodRegistry.registerService(service);
-    }
+      for (var method : service.bindService().getMethods()) {
+        var descriptor = method.getMethodDescriptor();
+        String methodFullName = descriptor.getFullMethodName();
+        registry.put(methodFullName, descriptor);
+        String routePath = "/" + methodFullName;
 
-    BindableService reflectionService = ProtoReflectionServiceV1.newInstance();
-    builder.addService(reflectionService);
-    methodRegistry.registerService(reflectionService);
+        //
+        app.post(
+            routePath,
+            ctx -> {
+              throw new IllegalStateException(
+                  "gRPC request reached the standard HTTP router for path: "
+                      + routePath
+                      + ". "
+                      + "This means the native gRPC server interceptor was bypassed. "
+                      + "Ensure you are running Jetty, Netty, or Undertow with HTTP/2 enabled, "
+                      + "and that the GrpcProcessor SPI is correctly loaded.");
+            });
+      }
+    }
 
     this.grpcServer = builder.build().start();
 
+    // KEEP .directExecutor() here!
+    // This ensures that when the background gRPC worker finishes, it instantly pushes
+    // the response back to Undertow/Netty without wasting time on another thread hop.
     var channel = InProcessChannelBuilder.forName(serverName).directExecutor().build();
+    var services = app.getServices();
+    var bridge = new DefaultGrpcProcessor(channel, registry);
 
-    var bridge = new UnifiedGrpcBridge(channel, methodRegistry);
     // Register it in the Service Registry so the server layer can find it
-    app.getServices().put(UnifiedGrpcBridge.class, bridge);
-
-    app.getServices().put(GrpcProcessor.class, bridge);
-
-    // Mount the bridge.
-    // app.post("/*", bridge);
+    services.put(DefaultGrpcProcessor.class, bridge);
+    services.put(GrpcProcessor.class, bridge);
 
-    app.onStop(
-        () -> {
-          channel.shutdownNow();
-          grpcServer.shutdownNow();
-        });
+    app.onStop(channel::shutdownNow);
+    app.onStop(grpcServer::shutdownNow);
   }
 }

modules/jooby-grpc/src/main/java/io/jooby/grpc/GrpcModule.java

@@ -3,7 +3,7 @@
  * Apache License Version 2.0 https://jooby.io/LICENSE.txt
  * Copyright 2014 Edgar Espina
  */
-package io.jooby.grpc;
+package io.jooby.internal.grpc;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -16,8 +16,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
 import io.grpc.CallOptions;
-import io.grpc.ClientCall;
 import io.grpc.ManagedChannel;
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
@@ -27,7 +27,7 @@
 import io.jooby.GrpcExchange;
 import io.jooby.GrpcProcessor;
 
-public class UnifiedGrpcBridge implements GrpcProcessor {
+public class DefaultGrpcProcessor implements GrpcProcessor {
 
   // Minimal Marshaller to pass raw bytes through the bridge
   private static class RawMarshaller implements MethodDescriptor.Marshaller<byte[]> {
@@ -48,24 +48,39 @@ public byte[] parse(InputStream stream) {
 
   private final Logger log = LoggerFactory.getLogger(getClass());
   private final ManagedChannel channel;
-  private final GrpcMethodRegistry methodRegistry;
+  private final Map<String, MethodDescriptor<?, ?>> registry;
 
-  public UnifiedGrpcBridge(ManagedChannel channel, GrpcMethodRegistry methodRegistry) {
+  public DefaultGrpcProcessor(
+      ManagedChannel channel, Map<String, MethodDescriptor<?, ?>> registry) {
     this.channel = channel;
-    this.methodRegistry = methodRegistry;
+    this.registry = registry;
   }
 
   @Override
-  public Flow.Subscriber<ByteBuffer> process(GrpcExchange exchange) {
+  public boolean isGrpcMethod(String path) {
+    // gRPC paths typically come in as "/package.Service/Method"
+    // Our registry stores them as "package.Service/Method"
+    String methodName = path.startsWith("/") ? path.substring(1) : path;
+
+    // Quick O(1) hash map lookup
+    return registry.get(methodName) != null;
+  }
+
+  @Override
+  public @NonNull Flow.Subscriber<ByteBuffer> process(@NonNull GrpcExchange exchange) {
     // Route paths: /{package.Service}/{Method}
     String path = exchange.getRequestPath();
     // Remove the leading slash to match the gRPC method registry format
-    var descriptor = methodRegistry.get(path.substring(1));
+    var descriptor = registry.get(path.substring(1));
 
     if (descriptor == null) {
-      log.warn("Method not found in bridge registry: {}", path);
-      exchange.close(Status.UNIMPLEMENTED.getCode().value(), "Method not found");
-      return null;
+      // MUST never occur, it is guarded by {@link #isGrpcMethod}
+      throw new IllegalStateException(
+          "Unregistered gRPC method: '"
+              + path
+              + "'. "
+              + "This request bypassed the GrpcProcessor.isGrpcMethod() guard, "
+              + "indicating a bug or misconfiguration in the native server interceptor.");
     }
 
     var method =
@@ -76,25 +91,24 @@ public Flow.Subscriber<ByteBuffer> process(GrpcExchange exchange) {
             .setResponseMarshaller(new RawMarshaller())
             .build();
 
-    CallOptions callOptions = extractCallOptions(exchange);
-    io.grpc.Metadata metadata = extractMetadata(exchange);
+    var callOptions = extractCallOptions(exchange);
+    var metadata = extractMetadata(exchange);
 
-    io.grpc.Channel interceptedChannel =
+    var interceptedChannel =
         io.grpc.ClientInterceptors.intercept(
             channel, io.grpc.stub.MetadataUtils.newAttachHeadersInterceptor(metadata));
 
-    ClientCall<byte[], byte[]> call = interceptedChannel.newCall(method, callOptions);
-    AtomicBoolean isFinished = new AtomicBoolean(false);
+    var call = interceptedChannel.newCall(method, callOptions);
+    var isFinished = new AtomicBoolean(false);
 
     boolean isUnaryOrServerStreaming =
         method.getType() == MethodDescriptor.MethodType.UNARY
             || method.getType() == MethodDescriptor.MethodType.SERVER_STREAMING;
 
-    // 1. Create the effectively final bridge
-    GrpcRequestBridge requestBridge = new GrpcRequestBridge(call, method.getType());
+    var requestBridge = new GrpcRequestBridge(call, method.getType());
 
-    ClientResponseObserver<byte[], byte[]> responseObserver =
-        new ClientResponseObserver<>() {
+    var responseObserver =
+        new ClientResponseObserver<byte[], byte[]>() {
 
           @Override
           public void beforeStart(ClientCallStreamObserver<byte[]> requestStream) {
@@ -158,10 +172,10 @@ private CallOptions extractCallOptions(GrpcExchange exchange) {
     }
 
     try {
-      char unit = timeout.charAt(timeout.length() - 1);
-      long value = Long.parseLong(timeout.substring(0, timeout.length() - 1));
+      var unit = timeout.charAt(timeout.length() - 1);
+      var value = Long.parseLong(timeout.substring(0, timeout.length() - 1));
 
-      java.util.concurrent.TimeUnit timeUnit =
+      var timeUnit =
           switch (unit) {
             case 'H' -> java.util.concurrent.TimeUnit.HOURS;
             case 'M' -> java.util.concurrent.TimeUnit.MINUTES;
@@ -184,10 +198,10 @@ private CallOptions extractCallOptions(GrpcExchange exchange) {
 
   /** Maps standard HTTP headers from the GrpcExchange into gRPC Metadata. */
   private io.grpc.Metadata extractMetadata(GrpcExchange exchange) {
-    io.grpc.Metadata metadata = new io.grpc.Metadata();
+    var metadata = new io.grpc.Metadata();
 
-    for (Map.Entry<String, String> header : exchange.getHeaders().entrySet()) {
-      String key = header.getKey().toLowerCase();
+    for (var header : exchange.getHeaders().entrySet()) {
+      var key = header.getKey().toLowerCase();
 
       if (key.startsWith(":")
           || key.startsWith("grpc-")
@@ -212,7 +226,7 @@ private io.grpc.Metadata extractMetadata(GrpcExchange exchange) {
 
   /** Prepends the 5-byte gRPC header and returns a ready-to-write ByteBuffer. */
   private ByteBuffer addGrpcHeader(byte[] payload) {
-    ByteBuffer buffer = ByteBuffer.allocate(5 + payload.length);
+    var buffer = ByteBuffer.allocate(5 + payload.length);
     buffer.put((byte) 0); // Compressed flag (0 = none)
     buffer.putInt(payload.length);
     buffer.put(payload);

…ava/io/jooby/grpc/UnifiedGrpcBridge.java …/internal/grpc/DefaultGrpcProcessor.javamodules/jooby-grpc/src/main/java/io/jooby/grpc/UnifiedGrpcBridge.java renamed to modules/jooby-grpc/src/main/java/io/jooby/internal/grpc/DefaultGrpcProcessor.java modules/jooby-grpc/src/main/java/io/jooby/grpc/UnifiedGrpcBridge.java renamed to modules/jooby-grpc/src/main/java/io/jooby/internal/grpc/DefaultGrpcProcessor.java

@@ -3,7 +3,7 @@
  * Apache License Version 2.0 https://jooby.io/LICENSE.txt
  * Copyright 2014 Edgar Espina
  */
-package io.jooby.grpc;
+package io.jooby.internal.grpc;
 
 import java.nio.ByteBuffer;
 import java.util.function.Consumer;

…ain/java/io/jooby/grpc/GrpcDeframer.java …io/jooby/internal/grpc/GrpcDeframer.javamodules/jooby-grpc/src/main/java/io/jooby/grpc/GrpcDeframer.java renamed to modules/jooby-grpc/src/main/java/io/jooby/internal/grpc/GrpcDeframer.java modules/jooby-grpc/src/main/java/io/jooby/grpc/GrpcDeframer.java renamed to modules/jooby-grpc/src/main/java/io/jooby/internal/grpc/GrpcDeframer.java

@@ -3,7 +3,7 @@
  * Apache License Version 2.0 https://jooby.io/LICENSE.txt
  * Copyright 2014 Edgar Espina
  */
-package io.jooby.grpc;
+package io.jooby.internal.grpc;
 
 import java.nio.ByteBuffer;
 import java.util.concurrent.Flow.Subscriber;