GrpcServiceBridge should only use client accepted encodings.

Motivation:

The GrpcServiceBridge implementation does not check the encodings accepted by the client and can use an encoding that the client would not support.

Changes:

- Add API to properly check client accepted encodings.
- Modify GrpcServiceBridge implementation to check the client accepted encodings.

Author: vietj

vertx-grpc-common/src/main/java/io/vertx/grpc/common/GrpcWriteStream.java

@@ -17,9 +17,12 @@ public interface GrpcWriteStream<T> extends WriteStream<T> {
   MultiMap headers();
 
   /**
-   * Set the stream encoding, e.g. {@code identity} or {@code gzip}.
+   * <p>Set the stream encoding, e.g. {@code identity} or {@code gzip},</p>
    *
-   * It must be called before sending any message, otherwise {@code identity} will be used.
+   * <ul>
+   *   <li>The encoding must be set before sending any message, otherwise {@code identity} will be used.</li>
+   *   <li>The encoding should also match the opposite endpoint expectations.</li>
+   * </ul>
    *
    * @param encoding the target message encoding
    * @return a reference to this, so the API can be used fluently

vertx-grpc-docs/src/main/asciidoc/server.adoc

@@ -316,7 +316,10 @@ Anemic JSON is also supported with Vert.x `JsonObject`
 
 === Compression
 
-You can compress response messages by setting the response encoding *prior* before sending any message
+You can compress response messages by setting the response encoding *prior* before sending any message.
+
+Before setting the encoding, you should use {@link io.vertx.grpc.server.GrpcServerResponse#acceptedEncodings()} to ensure
+the client RPC supports the encoding algorithm.
 
 [source,java]
 ----

vertx-grpc-docs/src/main/java/examples/GrpcServerExamples.java

@@ -170,7 +170,9 @@ public void anemicJson(GrpcServer server) {
   }
 
   public void responseCompression(GrpcServerResponse<Empty, Item> response) {
-    response.encoding("gzip");
+    if (response.acceptedEncodings().contains("gzip")) {
+      response.encoding("gzip");
+    }
 
     // Write items after encoding has been defined
     response.write(Item.newBuilder().setValue("item-1").build());

vertx-grpc-server/src/main/java/io/vertx/grpc/server/GrpcServerResponse.java

@@ -21,6 +21,8 @@
 import io.vertx.grpc.common.GrpcWriteStream;
 import io.vertx.grpc.common.WireFormat;
 
+import java.util.Set;
+
 @VertxGen
 public interface GrpcServerResponse<Req, Resp> extends GrpcWriteStream<Resp> {
 
@@ -48,6 +50,13 @@ public interface GrpcServerResponse<Req, Resp> extends GrpcWriteStream<Resp> {
   @Fluent
   GrpcServerResponse<Req, Resp> format(WireFormat format);
 
+  /**
+   * @return the set of accepted encodings sent by the client, note that {@code identity} should not be part of this set.
+   *         This can be used to set the response {@link #encoding(String) encoding} to ensure the client will accept
+   *         the encoding. This is a glorified wrapper for the {@code grpc-accept-encoding} header.
+   */
+  Set<String> acceptedEncodings();
+
   /**
    * @return the {@link MultiMap} to write metadata trailers
    */

vertx-grpc-server/src/main/java/io/vertx/grpc/server/impl/GrpcServerResponseImpl.java

@@ -26,19 +26,23 @@
 import io.vertx.grpc.server.GrpcServerResponse;
 import io.vertx.grpc.server.StatusException;
 
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
+import java.util.regex.Pattern;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
 public abstract class GrpcServerResponseImpl<Req, Resp> extends GrpcWriteStreamBase<GrpcServerResponseImpl<Req, Resp>, Resp> implements GrpcServerResponse<Req, Resp> {
 
+  private static final Pattern COMMA_SEPARATOR = Pattern.compile(" *, *");
+  private static final Set<String> GZIP_ACCEPT_ENCODING = Collections.singleton("gzip");
+
   private final GrpcServerRequestImpl<Req, Resp> request;
   private final HttpServerResponse httpResponse;
   private GrpcStatus status = GrpcStatus.OK;
   private String statusMessage;
   private boolean trailersOnly;
+  private Set<String> acceptedEncodings;
 
   public GrpcServerResponseImpl(ContextInternal context,
                                 GrpcServerRequestImpl<Req, Resp> request,
@@ -98,6 +102,27 @@ public GrpcStatus status() {
     return status;
   }
 
+  @Override
+  public Set<String> acceptedEncodings() {
+    if (acceptedEncodings == null) {
+      String acceptEncodingHeader = request.headers().get("grpc-accept-encoding");
+      if (acceptEncodingHeader != null) {
+        if (acceptEncodingHeader.equals("gzip")) {
+          acceptedEncodings = GZIP_ACCEPT_ENCODING;
+        } else {
+          acceptedEncodings = new HashSet<>(2);
+          String[] encodings = COMMA_SEPARATOR.split(acceptEncodingHeader);
+          for (String encoding : encodings) {
+            acceptedEncodings.add(encoding.trim());
+          }
+        }
+      } else {
+        acceptedEncodings = Collections.emptySet();
+      }
+    }
+    return acceptedEncodings;
+  }
+
   protected boolean sendCancel() {
     if (!isTrailersSent()) {
       status(GrpcStatus.CANCELLED);

vertx-grpc-server/src/test/java/io/vertx/tests/server/ServerRequestTest.java

@@ -50,21 +50,23 @@
 public class ServerRequestTest extends ServerTest {
 
   @Override
-  protected void testUnary(TestContext should, String requestEncoding, String responseEncoding) {
+  protected void testUnary(TestContext should, String requestEncoding, String responseEncoding, DecompressorRegistry decompressors) {
     startServer(GrpcServer.server(vertx).callHandler(UNARY, call -> {
       call.handler(helloRequest -> {
         Reply helloReply = Reply.newBuilder().setMessage("Hello " + helloRequest.getName()).build();
         if (!requestEncoding.equals("identity")) {
           should.assertEquals(requestEncoding, call.encoding());
         }
         GrpcServerResponse<Request, Reply> response = call.response();
+        if (response.acceptedEncodings().contains(responseEncoding)) {
+          response.encoding(responseEncoding);
+        }
         response
-          .encoding(responseEncoding)
           .end(helloReply);
       });
     }));
 
-    super.testUnary(should, requestEncoding, responseEncoding);
+    super.testUnary(should, requestEncoding, responseEncoding, decompressors);
   }
 
   @Test

vertx-grpc-server/src/test/java/io/vertx/tests/server/ServerTest.java

@@ -10,18 +10,7 @@
  */
 package io.vertx.tests.server;
 
-import io.grpc.CallOptions;
-import io.grpc.Channel;
-import io.grpc.ClientCall;
-import io.grpc.ClientInterceptor;
-import io.grpc.ClientInterceptors;
-import io.grpc.ForwardingClientCall;
-import io.grpc.ForwardingClientCallListener;
-import io.grpc.ManagedChannelBuilder;
-import io.grpc.Metadata;
-import io.grpc.MethodDescriptor;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
+import io.grpc.*;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import io.vertx.core.MultiMap;
@@ -38,6 +27,8 @@
 import io.vertx.tests.common.grpc.*;
 import org.junit.Test;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -54,21 +45,44 @@ public abstract class ServerTest extends ServerTestBase {
 
   @Test
   public void testUnary(TestContext should) {
-    testUnary(should, "identity", "identity");
+    testUnary(should, "identity", "identity", DecompressorRegistry.getDefaultInstance());
   }
 
   @Test
   public void testUnaryDecompression(TestContext should) {
-    testUnary(should, "gzip", "identity");
+    testUnary(should, "gzip", "identity", DecompressorRegistry.getDefaultInstance());
   }
 
   @Test
   public void testUnaryCompression(TestContext should) {
-    testUnary(should, "identity", "gzip");
+    testUnary(should, "identity", "gzip", DecompressorRegistry.getDefaultInstance());
   }
 
-  protected void testUnary(TestContext should, String requestEncoding, String responseEncoding) {
+  @Test
+  public void testUnaryCompressionWithUnsupportedEncoding(TestContext should) {
+    testUnary(should, "identity", "gzip", DecompressorRegistry.emptyInstance().with(Codec.Identity.NONE, false));
+  }
+
+  @Test
+  public void testUnaryCompressionWithMultipleValues(TestContext should) {
+    DecompressorRegistry registry = DecompressorRegistry.getDefaultInstance().with(new Decompressor() {
+      @Override
+      public String getMessageEncoding() {
+        return "custom";
+      }
+      @Override
+      public InputStream decompress(InputStream is) throws IOException {
+        return is;
+      }
+    }, true);
+    testUnary(should, "identity", "gzip", registry);
+  }
+
+  protected void testUnary(TestContext should, String requestEncoding, String responseEncoding, DecompressorRegistry decompressors) {
+
+
     channel = ManagedChannelBuilder.forAddress("localhost", port)
+      .decompressorRegistry(decompressors)
       .usePlaintext()
       .build();
 
@@ -94,7 +108,7 @@ public void onHeaders(Metadata headers) {
     Request request = Request.newBuilder().setName("Julien").build();
     Reply res = stub.unary(request);
     should.assertEquals("Hello Julien", res.getMessage());
-    if (!responseEncoding.equals("identity")) {
+    if (!responseEncoding.equals("identity") && decompressors.lookupDecompressor(responseEncoding) != null) {
       should.assertEquals(responseEncoding, responseGrpcEncoding.get());
     }
   }

vertx-grpcio-server/src/main/java/io/vertx/grpcio/server/impl/GrpcIoServiceBridgeImpl.java

@@ -268,7 +268,10 @@ public MethodDescriptor<Req, Resp> getMethodDescriptor() {
     @Override
     public void setCompression(String encoding) {
       compressor = CompressorRegistry.getDefaultInstance().lookupCompressor(encoding);
-      req.response().encoding(encoding);
+      GrpcServerResponse<Req, Resp> response = req.response();
+      if (response.acceptedEncodings().contains(encoding)) {
+        response.encoding(encoding);
+      }
     }
 
     @Override

vertx-grpcio-server/src/test/java/io/vertx/tests/server/ServerBridgeTest.java

@@ -35,12 +35,12 @@
 public class ServerBridgeTest extends ServerTest {
 
   @Override
-  protected void testUnary(TestContext should, String requestEncoding, String responseEncoding) {
+  protected void testUnary(TestContext should, String requestEncoding, String responseEncoding, DecompressorRegistry decompressors) {
     TestServiceGrpc.TestServiceImplBase impl = new TestServiceGrpc.TestServiceImplBase() {
       @Override
       public void unary(Request request, StreamObserver<Reply> responseObserver) {
         if (!responseEncoding.equals("identity")) {
-          ((ServerCallStreamObserver<?>)responseObserver).setCompression("gzip");
+          ((ServerCallStreamObserver<?>)responseObserver).setCompression(responseEncoding);
         }
         if (!requestEncoding.equals("identity")) {
           // No way to check the request encoding with the API
@@ -55,7 +55,7 @@ public void unary(Request request, StreamObserver<Reply> responseObserver) {
     serverStub.bind(server);
     startServer(server);
 
-    super.testUnary(should, requestEncoding, responseEncoding);
+    super.testUnary(should, requestEncoding, responseEncoding, decompressors);
   }
 
   @Test
@@ -123,7 +123,7 @@ public void onComplete() {
     serverStub.bind(server);
     startServer(server);
 
-    super.testUnary(should, "identity", "identity");
+    super.testUnary(should, "identity", "identity", DecompressorRegistry.getDefaultInstance());
   }
 
   @Test