decompressedMessageTooLong

Author: panchenko

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -330,8 +330,17 @@ private void messagesAvailableInternal(final MessageProducer producer) {
       InputStream message;
       try {
         while ((message = producer.next()) != null) {
+          ReqT parsedMessage;
           try {
-            listener.onMessage(call.method.parseRequest(message));
+            parsedMessage = call.method.parseRequest(message);
+          } catch (StatusRuntimeException e) {
+            GrpcUtil.closeQuietly(message);
+            call.cancelled = true;
+            call.close(e.getStatus(), new Metadata());
+            return;
+          }
+          try {
+            listener.onMessage(parsedMessage);
           } catch (Throwable t) {
             GrpcUtil.closeQuietly(message);
             throw t;

interop-testing/src/main/java/io/grpc/testing/integration/AbstractInteropTest.java

@@ -2030,7 +2030,7 @@ private void assertPayload(Payload expected, Payload actual) {
     }
   }
 
-  private static void assertCodeEquals(Status.Code expected, Status actual) {
+  protected static void assertCodeEquals(Status.Code expected, Status actual) {
     assertWithMessage("Unexpected status: %s", actual).that(actual.getCode()).isEqualTo(expected);
   }
 

interop-testing/src/test/java/io/grpc/testing/integration/TransportCompressionTest.java

@@ -17,6 +17,7 @@
 package io.grpc.testing.integration;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import com.google.protobuf.ByteString;
@@ -37,6 +38,8 @@
 import io.grpc.ServerCall.Listener;
 import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptor;
+import io.grpc.Status.Code;
+import io.grpc.StatusRuntimeException;
 import io.grpc.internal.GrpcUtil;
 import io.grpc.netty.InternalNettyChannelBuilder;
 import io.grpc.netty.InternalNettyServerBuilder;
@@ -53,7 +56,10 @@
 import java.io.OutputStream;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
@@ -84,10 +90,16 @@ public static void registerCompressors() {
     compressors.register(Codec.Identity.NONE);
   }
 
+  @Rule
+  public final TestName currentTest = new TestName();
+
   @Override
   protected ServerBuilder<?> getServerBuilder() {
     NettyServerBuilder builder = NettyServerBuilder.forPort(0, InsecureServerCredentials.create())
-        .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
+        .maxInboundMessageSize(
+            DECOMPRESSED_MESSAGE_TOO_LONG_METHOD_NAME.equals(currentTest.getMethodName())
+                ? 1000
+                : AbstractInteropTest.MAX_MESSAGE_SIZE)
         .compressorRegistry(compressors)
         .decompressorRegistry(decompressors)
         .intercept(new ServerInterceptor() {
@@ -126,6 +138,22 @@ public void compresses() {
     assertTrue(FZIPPER.anyWritten);
   }
 
+  private static final String DECOMPRESSED_MESSAGE_TOO_LONG_METHOD_NAME = "decompressedMessageTooLong";
+
+  @Test
+//  @Ignore("WIP")
+  public void decompressedMessageTooLong() {
+    assertEquals(DECOMPRESSED_MESSAGE_TOO_LONG_METHOD_NAME, currentTest.getMethodName());
+    final SimpleRequest bigRequest = SimpleRequest.newBuilder()
+        .setPayload(Payload.newBuilder().setBody(ByteString.copyFrom(new byte[271828])))
+        .build();
+    StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
+        () -> blockingStub.withCompression("gzip").unaryCall(bigRequest));
+    assertCodeEquals(Code.RESOURCE_EXHAUSTED, e.getStatus());
+    assertEquals("Decompressed gRPC message exceeds maximum size 1000",
+        e.getStatus().getDescription());
+  }
+
   @Override
   protected NettyChannelBuilder createChannelBuilder() {
     NettyChannelBuilder builder = NettyChannelBuilder.forAddress(getListenAddress())