Add proto transcoding option for DelegatingHttpJsonTranscodingService (line#6670)

Motivation

`DelegatingHttpJsonTranscodingService` transcodes incoming HTTP/JSON
requests into gRPC requests and forwards them to a delegate gRPC
service. Previously, the transcoded gRPC request was always serialized
as JSON (`application/grpc+json`) before being sent to the delegate.

However, when the delegate only supports `application/grpc+proto` (e.g.,
a standard gRPC service that does not have JSON marshalling enabled, or
a remote proxy upstream that expects proto wire format), the transcoded
request would be rejected or fail to deserialize correctly.

Modifications

- Added a `transcodedGrpcSerializationFormat(SerializationFormat)`
option to `DelegatingHttpJsonTranscodingServiceBuilder` (and the
internal `HttpJsonTranscoderBuilder`) that allows users to choose
between `GrpcSerializationFormats.JSON` (default, preserving existing
behaviour) and `GrpcSerializationFormats.PROTO`.
- When `PROTO` is selected, `HttpJsonTranscoder` converts the JSON
request body to a protobuf binary payload before framing and forwarding
to the delegate, and converts the protobuf binary response body back to
JSON before returning it to the HTTP client.
- Introduced a `JsonProtoMarshaller` helper inside `HttpJsonTranscoder`
that owns a pair of `GrpcMessageMarshaller` instances (one JSON, one
proto) and provides `jsonToProto` / `protoToJson` conversion using them.
- Refactored `GrpcMessageMarshaller` to expose public
`deserializeRequest(InputStream)`, `deserializeRequest(ByteBuf,
boolean)`, `deserializeResponse(InputStream)`, and
`deserializeResponse(ByteBuf, boolean)` overloads, extracting the
previously inlined per-type dispatch logic into dedicated methods to
support the new call sites.
- Added `grpcMethodDescriptor` and `jsonMarshaller` fields to
`HttpJsonTranscoderBuilder.HttpJsonGrpcMethod` so that the marshaller
pair can be constructed when proto transcoding is configured.
- Added a test case `shouldProxyHttpJsonRequestWithProtoUpstream` in
`DelegatingHttpJsonTranscodingServiceTest` to verify end-to-end
HTTP/JSON → proto → upstream → proto → JSON round-trip.
- Updated `DelegatingHttpJsonTranscodingDocServiceTest` to exercise the
new option.

Result

Users can now configure `DelegatingHttpJsonTranscodingService` to
forward transcoded requests using protobuf binary encoding to the
delegate, enabling interoperability with upstream gRPC services that
only accept `application/grpc+proto`.

```java
DelegatingHttpJsonTranscodingService.builder(grpcService)
    .serviceDescriptors(MyServiceGrpc.getServiceDescriptor())
    .transcodedGrpcSerializationFormat(GrpcSerializationFormats.PROTO)
    .build();
```

Author: jrhee17

grpc/src/main/java/com/linecorp/armeria/internal/common/grpc/GrpcMessageMarshaller.java

@@ -112,31 +112,19 @@ public ByteBuf serializeRequest(I message) throws IOException {
     }
 
     public I deserializeRequest(DeframedMessage message, boolean grpcWebText) throws IOException {
-        InputStream messageStream = message.stream();
+        final InputStream messageStream = message.stream();
         final ByteBuf buf = message.buf();
         if (buf != null) {
-            try {
-                switch (requestType) {
-                    case PROTOBUF:
-                        final PrototypeMarshaller<I> marshaller = (PrototypeMarshaller<I>) requestMarshaller;
-                        // PrototypeMarshaller<I>.getMessagePrototype will always parse to I
-                        @SuppressWarnings("unchecked")
-                        final I msg = (I) deserializeProto(marshaller, buf);
-                        return msg;
-                    default:
-                        // Fallback to using the method's stream marshaller.
-                        messageStream = new ByteBufInputStream(buf.retain(), true);
-                        break;
-                }
-            } finally {
-                if (!unsafeWrapDeserializedBuffer || grpcWebText) {
-                    buf.release();
-                }
-            }
+            return deserializeRequest(buf, grpcWebText);
         }
 
         assert messageStream != null;
-        try (InputStream msg = messageStream) {
+        return deserializeRequest(messageStream);
+    }
+
+    public I deserializeRequest(InputStream message) throws IOException {
+        requireNonNull(message, "message");
+        try (InputStream msg = message) {
             if (isProto) {
                 return method.parseRequest(msg);
             } else {
@@ -146,6 +134,29 @@ public I deserializeRequest(DeframedMessage message, boolean grpcWebText) throws
         }
     }
 
+    public I deserializeRequest(ByteBuf buf, boolean grpcWebText) throws IOException {
+        requireNonNull(buf, "buf");
+        try {
+            switch (requestType) {
+                case PROTOBUF:
+                    final PrototypeMarshaller<I> marshaller = (PrototypeMarshaller<I>) requestMarshaller;
+                    // PrototypeMarshaller<I>.getMessagePrototype will always parse to I
+                    @SuppressWarnings("unchecked")
+                    final I msg = (I) deserializeProto(marshaller, buf);
+                    return msg;
+                default:
+                    // Fallback to using the method's stream marshaller.
+                    try (InputStream messageStream = new ByteBufInputStream(buf.retain(), true)) {
+                        return deserializeRequest(messageStream);
+                    }
+            }
+        } finally {
+            if (!unsafeWrapDeserializedBuffer || grpcWebText) {
+                buf.release();
+            }
+        }
+    }
+
     public ByteBuf serializeResponse(O message) throws IOException {
         switch (responseType) {
             case PROTOBUF:
@@ -170,32 +181,19 @@ public ByteBuf serializeResponse(O message) throws IOException {
     }
 
     public O deserializeResponse(DeframedMessage message, boolean grpcWebText) throws IOException {
-        InputStream messageStream = message.stream();
+        final InputStream messageStream = message.stream();
         final ByteBuf buf = message.buf();
         if (buf != null) {
-            try {
-                switch (responseType) {
-                    case PROTOBUF:
-                        final PrototypeMarshaller<O> marshaller =
-                                (PrototypeMarshaller<O>) method.getResponseMarshaller();
-                        // PrototypeMarshaller<I>.getMessagePrototype will always parse to I
-                        @SuppressWarnings("unchecked")
-                        final O msg = (O) deserializeProto(marshaller, buf);
-                        return msg;
-                    default:
-                        // Fallback to using the method's stream marshaller.
-                        messageStream = new ByteBufInputStream(buf.retain(), true);
-                        break;
-                }
-            } finally {
-                if (!unsafeWrapDeserializedBuffer || grpcWebText) {
-                    buf.release();
-                }
-            }
+            return deserializeResponse(buf, grpcWebText);
         }
 
         assert messageStream != null;
-        try (InputStream msg = messageStream) {
+        return deserializeResponse(messageStream);
+    }
+
+    public O deserializeResponse(InputStream message) throws IOException {
+        requireNonNull(message, "message");
+        try (InputStream msg = message) {
             if (isProto) {
                 return method.parseResponse(msg);
             } else {
@@ -205,6 +203,30 @@ public O deserializeResponse(DeframedMessage message, boolean grpcWebText) throw
         }
     }
 
+    public O deserializeResponse(ByteBuf buf, boolean grpcWebText) throws IOException {
+        requireNonNull(buf, "buf");
+        try {
+            switch (responseType) {
+                case PROTOBUF:
+                    final PrototypeMarshaller<O> marshaller =
+                            (PrototypeMarshaller<O>) method.getResponseMarshaller();
+                    // PrototypeMarshaller<I>.getMessagePrototype will always parse to I
+                    @SuppressWarnings("unchecked")
+                    final O msg = (O) deserializeProto(marshaller, buf);
+                    return msg;
+                default:
+                    // Fallback to using the method's stream marshaller.
+                    try (InputStream messageStream = new ByteBufInputStream(buf.retain(), true)) {
+                        return deserializeResponse(messageStream);
+                    }
+            }
+        } finally {
+            if (!unsafeWrapDeserializedBuffer || grpcWebText) {
+                buf.release();
+            }
+        }
+    }
+
     private <T> ByteBuf serializeProto(PrototypeMarshaller<T> marshaller, Message message) throws IOException {
         if (isProto) {
             final int serializedSize = message.getSerializedSize();

grpc/src/main/java/com/linecorp/armeria/server/grpc/DelegatingHttpJsonTranscodingServiceBuilder.java

@@ -51,6 +51,7 @@ public final class DelegatingHttpJsonTranscodingServiceBuilder {
 
     private HttpJsonTranscodingOptions options = HttpJsonTranscodingOptions.of();
     private HttpService fallback = DEFAULT_FALLBACK;
+    private boolean protoSerialization = true;
 
     /**
      * Creates a new builder for the specified delegate.
@@ -77,6 +78,15 @@ public DelegatingHttpJsonTranscodingServiceBuilder options(HttpJsonTranscodingOp
         return this;
     }
 
+    /**
+     * Sets whether to use Protocol Buffers serialization for transcoded gRPC requests sent to the delegate.
+     * If {@code false}, JSON serialization is used instead. Defaults to {@code true}.
+     */
+    public DelegatingHttpJsonTranscodingServiceBuilder protoSerialization(boolean protoSerialization) {
+        this.protoSerialization = protoSerialization;
+        return this;
+    }
+
     /**
      * Adds the {@link ServiceDescriptor}s that define HTTP/JSON mappings.
      */
@@ -110,6 +120,7 @@ public DelegatingHttpJsonTranscodingService build() {
         final HttpJsonTranscoder transcoder =
                 new HttpJsonTranscoderBuilder()
                         .options(options)
+                        .protoSerialization(protoSerialization)
                         .serviceDescriptors(serviceDescriptors)
                         .build();
         if (transcoder == null) {

grpc/src/main/java/com/linecorp/armeria/server/grpc/GrpcServiceBuilder.java

@@ -1099,6 +1099,7 @@ public GrpcService build() {
             final HttpJsonTranscoder transcoder =
                     new HttpJsonTranscoderBuilder()
                             .options(httpJsonTranscodingOptions)
+                            .protoSerialization(false)
                             .serviceDefinitions(grpcService.services())
                             .build();
             if (transcoder != null) {

grpc/src/main/java/com/linecorp/armeria/server/grpc/HttpJsonTranscoder.java

@@ -22,6 +22,7 @@
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static java.util.Objects.requireNonNull;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.Base64;
@@ -69,12 +70,14 @@
 import com.linecorp.armeria.common.RequestHeaders;
 import com.linecorp.armeria.common.RequestHeadersBuilder;
 import com.linecorp.armeria.common.ResponseHeaders;
+import com.linecorp.armeria.common.SerializationFormat;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.grpc.GrpcSerializationFormats;
 import com.linecorp.armeria.common.grpc.protocol.GrpcHeaderNames;
 import com.linecorp.armeria.common.logging.RequestLogProperty;
 import com.linecorp.armeria.common.util.SafeCloseable;
 import com.linecorp.armeria.internal.common.JacksonUtil;
+import com.linecorp.armeria.internal.common.grpc.GrpcMessageMarshaller;
 import com.linecorp.armeria.internal.server.grpc.HttpEndpointSpecification;
 import com.linecorp.armeria.internal.server.grpc.HttpEndpointSpecification.Parameter;
 import com.linecorp.armeria.internal.server.grpc.HttpEndpointSupport;
@@ -90,6 +93,7 @@
 import com.linecorp.armeria.unsafe.PooledObjects;
 
 import io.grpc.Status;
+import io.netty.buffer.ByteBuf;
 import io.netty.util.AttributeKey;
 
 final class HttpJsonTranscoder implements HttpEndpointSupport {
@@ -146,12 +150,15 @@ private static String grpcPath(MethodDescriptor methodDescriptor) {
             AttributeKey.valueOf(FramedGrpcService.class, "HTTP_JSON_GRPC_METHOD_INFO");
 
     private final HttpJsonTranscodingOptions options;
+    private final boolean protoSerialization;
     private final Map<Route, TranscodingSpec> routeAndSpecs;
     private final Set<Route> routes;
 
     HttpJsonTranscoder(Map<Route, TranscodingSpec> routeAndSpecs,
-                       HttpJsonTranscodingOptions httpJsonTranscodingOptions) {
+                       HttpJsonTranscodingOptions httpJsonTranscodingOptions,
+                       boolean protoSerialization) {
         options = requireNonNull(httpJsonTranscodingOptions, "httpJsonTranscodingOptions");
+        this.protoSerialization = protoSerialization;
         this.routeAndSpecs = routeAndSpecs;
 
         final LinkedHashSet<Route> linkedHashSet = new LinkedHashSet<>(routeAndSpecs.size());
@@ -214,8 +221,10 @@ HttpResponse serve(ServiceRequestContext ctx, HttpRequest req,
                                    "gRPC encoding is not supported for non-framed requests.");
         }
 
+        final boolean useProto = protoSerialization;
         grpcHeaders.method(HttpMethod.POST)
-                   .contentType(GrpcSerializationFormats.JSON.mediaType());
+                   .contentType(useProto ? GrpcSerializationFormats.PROTO.mediaType()
+                                         : GrpcSerializationFormats.JSON.mediaType());
         grpcHeaders.path(grpcPath(spec.methodDescriptor));
         // All clients support no encoding, and we don't support gRPC encoding for non-framed requests, so just
         // clear the header if it's present.
@@ -244,10 +253,20 @@ HttpResponse serve(ServiceRequestContext ctx, HttpRequest req,
                             requestContent = convertToJson(ctx, clientRequest, spec);
                         }
 
+                        final JsonProtoMarshaller jsonProtoMarshaller =
+                                useProto ? new JsonProtoMarshaller(ctx, method) : null;
+                        final HttpData transcodedRequestContent;
+                        if (useProto) {
+                            transcodedRequestContent = convertToProto(requestContent, jsonProtoMarshaller);
+                        } else {
+                            transcodedRequestContent = requestContent;
+                        }
                         final ResponseHandler responseHandler =
-                                new HttpJsonResponseHandler(ctx, responseFuture, spec);
+                                new HttpJsonResponseHandler(ctx, responseFuture, spec,
+                                                            useProto, jsonProtoMarshaller);
                         UnframedGrpcSupport.frameAndServe(
-                                delegate, ctx, grpcHeaders.build(), requestContent, responseHandler);
+                                delegate, ctx, grpcHeaders.build(), transcodedRequestContent,
+                                responseHandler);
                     } catch (IllegalArgumentException iae) {
                         responseFuture.completeExceptionally(
                                 HttpStatusException.of(HttpStatus.BAD_REQUEST, iae));
@@ -261,7 +280,17 @@ HttpResponse serve(ServiceRequestContext ctx, HttpRequest req,
         return HttpResponse.of(responseFuture);
     }
 
-    private AggregatedHttpResponse convertResponse(TranscodingSpec spec, AggregatedHttpResponse httpResponse) {
+    private static AggregatedHttpResponse convertResponse(
+            TranscodingSpec spec, boolean useProto, AggregatedHttpResponse httpResponse,
+            @Nullable JsonProtoMarshaller jsonProtoMarshaller) throws Exception {
+        final AggregatedHttpResponse jsonResponse =
+                useProto ? convertProtoResponseToJson(httpResponse, jsonProtoMarshaller)
+                         : httpResponse;
+        return convertJsonResponse(spec, jsonResponse);
+    }
+
+    private static AggregatedHttpResponse convertJsonResponse(
+            TranscodingSpec spec, AggregatedHttpResponse httpResponse) {
         // Ignore the spec if the method is HttpBody. The response body is already in the correct format.
         if (HttpBody.getDescriptor().equals(spec.methodDescriptor.getOutputType())) {
             final HttpData data = httpResponse.content();
@@ -304,6 +333,60 @@ private AggregatedHttpResponse convertResponse(TranscodingSpec spec, AggregatedH
         }
     }
 
+    private static  HttpData convertToProto(HttpData requestContent,
+                                            @Nullable JsonProtoMarshaller jsonProtoMarshaller)
+            throws IOException {
+        final byte[] jsonBuf = requestContent.array();
+        assert jsonProtoMarshaller != null;
+        final ByteBuf protoBuf = jsonProtoMarshaller.jsonToProto(jsonBuf);
+        return HttpData.wrap(protoBuf);
+    }
+
+    private static AggregatedHttpResponse convertProtoResponseToJson(
+            AggregatedHttpResponse httpResponse,
+            @Nullable JsonProtoMarshaller jsonProtoMarshaller) throws Exception {
+        final HttpData data = httpResponse.content();
+        final ByteBuf protoBuf = data.byteBuf();
+        assert jsonProtoMarshaller != null;
+        final ByteBuf jsonBuf = jsonProtoMarshaller.protoToJson(protoBuf, true);
+        return AggregatedHttpResponse.of(httpResponse.headers(), HttpData.wrap(jsonBuf));
+    }
+
+    @SuppressWarnings("unchecked")
+    private static GrpcMessageMarshaller<Object, Object> newMessageMarshaller(
+            ServiceRequestContext ctx, SerializationFormat serializationFormat,
+            HttpJsonGrpcMethod method) {
+        return new GrpcMessageMarshaller<>(
+                ctx.alloc(),
+                serializationFormat,
+                (io.grpc.MethodDescriptor<Object, Object>) method.grpcMethodDescriptor,
+                method.jsonMarshaller,
+                false,
+                true);
+    }
+
+    private static final class JsonProtoMarshaller {
+        private final GrpcMessageMarshaller<Object, Object> json;
+        private final GrpcMessageMarshaller<Object, Object> proto;
+
+        JsonProtoMarshaller(ServiceRequestContext ctx, HttpJsonGrpcMethod method) {
+            json = newMessageMarshaller(ctx, GrpcSerializationFormats.JSON, method);
+            proto = newMessageMarshaller(ctx, GrpcSerializationFormats.PROTO, method);
+        }
+
+        ByteBuf jsonToProto(byte[] jsonBytes) throws IOException {
+            try (ByteArrayInputStream inputStream = new ByteArrayInputStream(jsonBytes)) {
+                final Object message = json.deserializeRequest(inputStream);
+                return proto.serializeRequest(message);
+            }
+        }
+
+        ByteBuf protoToJson(ByteBuf protoBuf, boolean grpcWebText) throws IOException {
+            final Object message = proto.deserializeResponse(protoBuf, grpcWebText);
+            return json.serializeResponse(message);
+        }
+    }
+
     private static HttpData convertToHttpBody(AggregatedHttpRequest request) throws IOException {
         final ObjectNode body = mapper.createObjectNode();
 
@@ -464,7 +547,7 @@ private static void setParametersToNode(ObjectNode root,
                 continue;
             }
 
-            if (field.javaType == JavaType.MESSAGE) {
+            if (field.type() == JavaType.MESSAGE) {
                 throw new IllegalArgumentException(
                         "Unsupported message type: " + field.descriptor.getFullName());
             }
@@ -759,13 +842,19 @@ private class HttpJsonResponseHandler implements ResponseHandler {
         private final ServiceRequestContext ctx;
         private final CompletableFuture<HttpResponse> responseFuture;
         private final TranscodingSpec spec;
+        private final boolean useProto;
+        @Nullable
+        private final JsonProtoMarshaller jsonProtoMarshaller;
 
         HttpJsonResponseHandler(ServiceRequestContext ctx,
                                 CompletableFuture<HttpResponse> responseFuture,
-                                TranscodingSpec spec) {
+                                TranscodingSpec spec, boolean useProto,
+                                @Nullable JsonProtoMarshaller jsonProtoMarshaller) {
             this.ctx = ctx;
             this.responseFuture = responseFuture;
             this.spec = spec;
+            this.useProto = useProto;
+            this.jsonProtoMarshaller = jsonProtoMarshaller;
         }
 
         @Override
@@ -786,7 +875,8 @@ public void handle(@Nullable AggregatedHttpResponse aggregatedResponse,
                     return;
                 }
                 assert aggregatedResponse != null;
-                final AggregatedHttpResponse convertedResponse = convertResponse(spec, aggregatedResponse);
+                final AggregatedHttpResponse convertedResponse =
+                        convertResponse(spec, useProto, aggregatedResponse, jsonProtoMarshaller);
                 responseFuture.complete(convertedResponse.toHttpResponse());
             } catch (Exception e) {
                 responseFuture.completeExceptionally(e);