#9870: updating author to ioanbsu <[email protected]>

Author: ioanbsu

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

@@ -184,15 +184,41 @@ 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

@@ -33,6 +33,7 @@
 import io.grpc.ServerCall.Listener;
 import io.grpc.internal.NoopServerCall;
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -425,6 +426,91 @@ public void onMessage(ReqT message) {
         order);
   }
 
+  /**
+   * Tests the {@link ServerInterceptors#useMarshalledMessages(ServerServiceDefinition, Marshaller, Marshaller)}.
+   * 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
+  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 new ByteArrayInputStream(value.getBytes());
+      }
+
+      @Override
+      public String parse(InputStream stream) {
+        requestFlowOrder.add("RequestParse");
+        try {
+          byte[] bytes = new byte[stream.available()];
+          stream.read(bytes);
+          return new String(bytes);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    };
+    final Marshaller<String> responseMarshaller = new Marshaller<String>() {
+      @Override
+      public InputStream stream(String value) {
+        requestFlowOrder.add("ResponseStream");
+        return new ByteArrayInputStream(value.getBytes());
+      }
+
+      @Override
+      public String parse(InputStream stream) {
+        requestFlowOrder.add("ResponseParse");
+        try {
+          byte[] bytes = new byte[stream.available()];
+          stream.read(bytes);
+          return new String(bytes);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    };
+    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) {