Updating ServerInterceptors.java to support different marshallers for Request and Response messages. (#9877)

Fixes #9870

Author: ioanbsu

api/src/main/java/io/grpc/ServerInterceptors.java

@@ -184,15 +184,40 @@ private static final class KnownLengthBufferedInputStream extends BufferedInputS
   public static <T> ServerServiceDefinition useMarshalledMessages(
       final ServerServiceDefinition serviceDef,
       final MethodDescriptor.Marshaller<T> marshaller) {
+    return useMarshalledMessages(serviceDef, marshaller, marshaller);
+  }
+
+  /**
+   * Create a new {@code ServerServiceDefinition} with {@link MethodDescriptor} for deserializing
+   * requests and separate {@link MethodDescriptor} for serializing responses. The {@code
+   * ServerCallHandler} created will automatically convert back to the original types for request
+   * and response before calling the existing {@code ServerCallHandler}.  Calling this method
+   * combined with the intercept methods will allow the developer to choose whether to intercept
+   * messages of ReqT/RespT, or the modeled types of their application. This can also be chained
+   * to allow for interceptors to handle messages as multiple different ReqT/RespT types within
+   * the chain if the added cost of serialization is not a concern.
+   *
+   * @param serviceDef         the sevice definition to add request and response marshallers to.
+   * @param requestMarshaller  request marshaller
+   * @param responseMarshaller response marshaller
+   * @param <ReqT>             the request payload type
+   * @param <RespT>            the response payload type.
+   * @return a wrapped version of {@code serviceDef} with the ReqT and RespT conversion applied.
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/9870")
+  public static <ReqT, RespT> ServerServiceDefinition useMarshalledMessages(
+      final ServerServiceDefinition serviceDef,
+      final MethodDescriptor.Marshaller<ReqT> requestMarshaller,
+      final MethodDescriptor.Marshaller<RespT> responseMarshaller) {
     List<ServerMethodDefinition<?, ?>> wrappedMethods =
         new ArrayList<>();
     List<MethodDescriptor<?, ?>> wrappedDescriptors =
         new ArrayList<>();
     // Wrap the descriptors
     for (final ServerMethodDefinition<?, ?> definition : serviceDef.getMethods()) {
       final MethodDescriptor<?, ?> originalMethodDescriptor = definition.getMethodDescriptor();
-      final MethodDescriptor<T, T> wrappedMethodDescriptor =
-          originalMethodDescriptor.toBuilder(marshaller, marshaller).build();
+      final MethodDescriptor<ReqT, RespT> wrappedMethodDescriptor =
+          originalMethodDescriptor.toBuilder(requestMarshaller, responseMarshaller).build();
       wrappedDescriptors.add(wrappedMethodDescriptor);
       wrappedMethods.add(wrapMethod(definition, wrappedMethodDescriptor));
     }

api/src/test/java/io/grpc/ServerInterceptorsTest.java

@@ -425,6 +425,80 @@ public void onMessage(ReqT message) {
         order);
   }
 
+  /**
+   * Tests the ServerInterceptors#useMarshalledMessages()} with two marshallers. Makes sure that
+   * on incoming request the request marshaller's stream method is called and on response the
+   * response marshaller's parse method is called
+   */
+  @Test
+  @SuppressWarnings("unchecked")
+  public void distinctMarshallerForRequestAndResponse() {
+    final List<String> requestFlowOrder = new ArrayList<>();
+
+    final Marshaller<String> requestMarshaller = new Marshaller<String>() {
+      @Override
+      public InputStream stream(String value) {
+        requestFlowOrder.add("RequestStream");
+        return null;
+      }
+
+      @Override
+      public String parse(InputStream stream) {
+        requestFlowOrder.add("RequestParse");
+        return null;
+      }
+    };
+    final Marshaller<String> responseMarshaller = new Marshaller<String>() {
+      @Override
+      public InputStream stream(String value) {
+        requestFlowOrder.add("ResponseStream");
+        return null;
+      }
+
+      @Override
+      public String parse(InputStream stream) {
+        requestFlowOrder.add("ResponseParse");
+        return null;
+      }
+    };
+    final Marshaller<Holder> dummyMarshaller = new Marshaller<Holder>() {
+      @Override
+      public InputStream stream(Holder value) {
+        return value.get();
+      }
+
+      @Override
+      public Holder parse(InputStream stream) {
+        return new Holder(stream);
+      }
+    };
+    ServerCallHandler<Holder, Holder> handler = (call, headers) -> new Listener<Holder>() {
+      @Override
+      public void onMessage(Holder message) {
+        requestFlowOrder.add("handler");
+        call.sendMessage(message);
+      }
+    };
+
+    MethodDescriptor<Holder, Holder> wrappedMethod = MethodDescriptor.<Holder, Holder>newBuilder()
+        .setType(MethodType.UNKNOWN)
+        .setFullMethodName("basic/wrapped")
+        .setRequestMarshaller(dummyMarshaller)
+        .setResponseMarshaller(dummyMarshaller)
+        .build();
+    ServerServiceDefinition serviceDef = ServerServiceDefinition.builder(
+            new ServiceDescriptor("basic", wrappedMethod))
+        .addMethod(wrappedMethod, handler).build();
+    ServerServiceDefinition intercepted = ServerInterceptors.useMarshalledMessages(serviceDef,
+        requestMarshaller, responseMarshaller);
+    ServerMethodDefinition<String, String> serverMethod =
+        (ServerMethodDefinition<String, String>) intercepted.getMethod("basic/wrapped");
+    ServerCall<String, String> serverCall = new NoopServerCall<>();
+    serverMethod.getServerCallHandler().startCall(serverCall, headers).onMessage("TestMessage");
+
+    assertEquals(Arrays.asList("RequestStream",  "handler", "ResponseParse"), requestFlowOrder);
+  }
+
   @SuppressWarnings("unchecked")
   private static ServerMethodDefinition<String, Integer> getSoleMethod(
       ServerServiceDefinition serviceDef) {