feat(transcoding): initial version of complex paths

zZHorizonZz · Apr 23, 2024 · 320a63f · 320a63f
1 parent f77df97
commit 320a63f
Show file tree

Hide file tree

Showing 11 changed files with 446 additions and 48 deletions.
diff --git a/extensions/grpc/api/src/main/java/io/quarkus/grpc/GrpcTranscodingDescriptor.java b/extensions/grpc/api/src/main/java/io/quarkus/grpc/GrpcTranscodingDescriptor.java
@@ -0,0 +1,21 @@
+package io.quarkus.grpc;
+
+public class GrpcTranscodingDescriptor<Req extends com.google.protobuf.Message, Resp extends com.google.protobuf.Message> {
+
+    private final GrpcTranscodingMarshaller<Req> requestMarshaller;
+    private final GrpcTranscodingMarshaller<Resp> responseMarshaller;
+
+    public GrpcTranscodingDescriptor(GrpcTranscodingMarshaller<Req> requestMarshaller,
+            GrpcTranscodingMarshaller<Resp> responseMarshaller) {
+        this.requestMarshaller = requestMarshaller;
+        this.responseMarshaller = responseMarshaller;
+    }
+
+    public GrpcTranscodingMarshaller<Req> getRequestMarshaller() {
+        return requestMarshaller;
+    }
+
+    public GrpcTranscodingMarshaller<Resp> getResponseMarshaller() {
+        return responseMarshaller;
+    }
+}
diff --git a/extensions/grpc/api/src/main/java/io/quarkus/grpc/MutinyTranscodingService.java b/extensions/grpc/api/src/main/java/io/quarkus/grpc/MutinyTranscodingService.java
@@ -6,7 +6,6 @@ public interface MutinyTranscodingService {
 
     String getGrpcServiceName();
 
-    <T extends Message> GrpcTranscodingMarshaller<T> findRequestMarshaller(String methodName);
-
-    <T extends Message> GrpcTranscodingMarshaller<T> findResponseMarshaller(String methodName);
+    <Req extends Message, Resp extends Message> GrpcTranscodingDescriptor<Req, Resp> findTranscodingDescriptor(
+            String methodName);
 }
diff --git a/extensions/grpc/protoc/src/main/resources/MutinyMarshalling.mustache b/extensions/grpc/protoc/src/main/resources/MutinyMarshalling.mustache
@@ -5,6 +5,8 @@ package {{packageName}};
 import jakarta.ws.rs.core.Response;
 import io.quarkus.grpc.GrpcService;
 import io.quarkus.grpc.MutinyTranscodingService;
+import io.quarkus.grpc.GrpcTranscodingDescriptor;
+import com.google.protobuf.Message;
 import com.google.protobuf.util.JsonFormat;
 import com.google.protobuf.InvalidProtocolBufferException;
 
@@ -19,40 +21,23 @@ public class {{serviceName}}Marshalling implements MutinyTranscodingService {
     }
 
     @Override
-    public <T extends com.google.protobuf.Message> io.quarkus.grpc.GrpcTranscodingMarshaller<T> findRequestMarshaller(String method) {
+    public <Req extends Message, Resp extends Message> GrpcTranscodingDescriptor<Req, Resp> findTranscodingDescriptor(String method) {
         switch (method) {
-            {{#unaryUnaryMethods}}
+        {{#unaryUnaryMethods}}
             case "{{methodName}}":
-                return (io.quarkus.grpc.GrpcTranscodingMarshaller<T>) {{methodName}}RequestMarshaller();
-            {{/unaryUnaryMethods}}
-            default:
-                throw new IllegalArgumentException("Unknown request method: " + method);
+            return (GrpcTranscodingDescriptor<Req, Resp>) {{methodName}}TranscodingDescriptor();
+        {{/unaryUnaryMethods}}
+        default:
+            throw new IllegalArgumentException("Unknown request method: " + method);
         }
     }
 
-    @Override
-    public <T extends com.google.protobuf.Message> io.quarkus.grpc.GrpcTranscodingMarshaller<T> findResponseMarshaller(String method) {
-        switch (method) {
-            {{#unaryUnaryMethods}}
-            case "{{methodName}}":
-                return (io.quarkus.grpc.GrpcTranscodingMarshaller<T>) {{methodName}}ResponseMarshaller();
-            {{/unaryUnaryMethods}}
-            default:
-                throw new IllegalArgumentException("Unknown response method: " + method);
-        }
-    }
-
-    {{#unaryUnaryMethods}}
-    @io.quarkus.grpc.GrpcTranscodingMethod(grpcMethodName = "{{methodName}}", httpMethod = "{{httpMethod}}", httpPath = "{{httpPath}}")
-    public io.quarkus.grpc.GrpcTranscodingMarshaller<{{inputType}}> {{methodName}}RequestMarshaller() {
-       return new io.quarkus.grpc.GrpcTranscodingMarshaller<{{inputType}}>({{inputType}}.getDefaultInstance());
-    }
-    {{/unaryUnaryMethods}}
-
     {{#unaryUnaryMethods}}
     @io.quarkus.grpc.GrpcTranscodingMethod(grpcMethodName = "{{methodName}}", httpMethod = "{{httpMethod}}", httpPath = "{{httpPath}}")
-    public io.quarkus.grpc.GrpcTranscodingMarshaller<{{outputType}}> {{methodName}}ResponseMarshaller() {
-       return new io.quarkus.grpc.GrpcTranscodingMarshaller<{{outputType}}>({{outputType}}.getDefaultInstance());
+    public io.quarkus.grpc.GrpcTranscodingDescriptor<{{inputType}}, {{outputType}}> {{methodName}}TranscodingDescriptor() {
+        return new io.quarkus.grpc.GrpcTranscodingDescriptor<{{inputType}}, {{outputType}}>(
+            new io.quarkus.grpc.GrpcTranscodingMarshaller<{{inputType}}>({{inputType}}.getDefaultInstance()),
+            new io.quarkus.grpc.GrpcTranscodingMarshaller<{{outputType}}>({{outputType}}.getDefaultInstance()));
     }
     {{/unaryUnaryMethods}}
 }
diff --git a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcTranscodingRecorder.java b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcTranscodingRecorder.java
@@ -15,7 +15,7 @@
 import io.grpc.ServerMethodDefinition;
 import io.grpc.ServerServiceDefinition;
 import io.quarkus.arc.Arc;
-import io.quarkus.grpc.GrpcTranscodingMarshaller;
+import io.quarkus.grpc.GrpcTranscodingDescriptor;
 import io.quarkus.grpc.MutinyTranscodingService;
 import io.quarkus.grpc.auth.GrpcSecurityInterceptor;
 import io.quarkus.grpc.runtime.transcoding.*;
@@ -78,7 +78,7 @@ public RuntimeValue<GrpcTranscodingServer> initializeMarshallingServer(RuntimeVa
 
                 mappedMethods.add(serviceDefinition);
 
-                Route route = routerSupplier.getValue().route(httpMethod, path).handler(ctx -> {
+                Route route = routerSupplier.getValue().route().handler(ctx -> {
                     if (securityPresent) {
                         GrpcSecurityInterceptor.propagateSecurityIdentityWithDuplicatedCtx(ctx);
                     }
@@ -115,11 +115,13 @@ private <Req extends Message, Resp extends Message> GrpcTranscodingMetadata<Req,
                 .substring(methodDescriptor.getFullMethodName().lastIndexOf("/") + 1);
         fullMethodName = Character.toLowerCase(fullMethodName.charAt(0)) + fullMethodName.substring(1);
 
-        GrpcTranscodingMarshaller<Req> requestMarshaller = transcodingService.findRequestMarshaller(fullMethodName);
-        GrpcTranscodingMarshaller<Resp> responseMarshaller = transcodingService.findResponseMarshaller(fullMethodName);
+        GrpcTranscodingDescriptor<Req, Resp> descriptor = transcodingService.findTranscodingDescriptor(fullMethodName);
 
-        return new GrpcTranscodingMetadata<>(transcodingMethod.getHttpMethodName(), fullMethodName, requestMarshaller,
-                responseMarshaller, methodDescriptor);
+        return new GrpcTranscodingMetadata<>(transcodingMethod.getHttpMethodName(),
+                fullMethodName,
+                descriptor.getRequestMarshaller(),
+                descriptor.getResponseMarshaller(),
+                methodDescriptor);
     }
 
     private List<GrpcTranscodingMethod> findTranscodingMethods(Map<String, List<GrpcTranscodingMethod>> transcodingMethods,

diff --git a/...rpc/runtime/src/main/java/io/quarkus/grpc/runtime/transcoding/GrpcTranscodingRequest.java b/...rpc/runtime/src/main/java/io/quarkus/grpc/runtime/transcoding/GrpcTranscodingRequest.java
@@ -84,7 +84,7 @@ public void init() {
         });
         queue.drainHandler(v -> stream.resume());
         queue.handler(msg -> {
-            if (msg == END_SENTINEL) {
+            if (msg == END_SENTINEL && httpRequest.bytesRead() != 0) {
                 handleEnd();
             } else {
                 handleMessage(msg);
@@ -267,4 +267,8 @@ public <R, C> Future<R> collecting(Collector<Req, C, R> collector) {
     public HttpConnection connection() {
         return httpRequest.connection();
     }
+
+    public HttpServerRequest httpRequest() {
+        return httpRequest;
+    }
 }
diff --git a/...pc/runtime/src/main/java/io/quarkus/grpc/runtime/transcoding/GrpcTranscodingResponse.java b/...pc/runtime/src/main/java/io/quarkus/grpc/runtime/transcoding/GrpcTranscodingResponse.java
@@ -152,11 +152,6 @@ private Future<Void> writeMessage(GrpcMessage message, boolean end) {
             throw new IllegalStateException();
         }
 
-        Buffer encoded = encode(message);
-        if (encoded == null) {
-            throw new IllegalStateException("The message is null");
-        }
-
         boolean trailersOnly = status != GrpcStatus.OK && !headersSent && end;
 
         MultiMap responseHeaders = httpResponse.headers();
@@ -169,7 +164,6 @@ private Future<Void> writeMessage(GrpcMessage message, boolean end) {
             }
 
             responseHeaders.set("content-type", "application/json");
-            responseHeaders.set("content-length", String.valueOf(encoded.length()));
         }
 
         if (end) {
@@ -200,11 +194,23 @@ private Future<Void> writeMessage(GrpcMessage message, boolean end) {
                 responseTrailers.remove("grpc-message");
             }
             if (message != null) {
+                Buffer encoded = encode(message);
+                if (encoded == null) {
+                    throw new IllegalStateException("The message is null");
+                }
+
+                responseHeaders.set("content-length", String.valueOf(encoded.length()));
                 return httpResponse.end(encoded);
             } else {
                 return httpResponse.end();
             }
         } else {
+            Buffer encoded = encode(message);
+            if (encoded == null) {
+                throw new IllegalStateException("The message is null");
+            }
+
+            responseHeaders.set("content-length", String.valueOf(encoded.length()));
             return httpResponse.write(encoded);
         }
     }