Merge pull request #43596 from cescoffier/grpc-server-message-size

quarkusio · Sep 30, 2024 · 4cab5df · 4cab5df
2 parents 351b9f9 + 93e6eac
commit 4cab5df
Show file tree

Hide file tree

Showing 4 changed files with 152 additions and 1 deletion.
diff --git a/...ns/grpc/deployment/src/test/java/io/quarkus/grpc/server/GrpcServerInboundMessageTest.java b/...ns/grpc/deployment/src/test/java/io/quarkus/grpc/server/GrpcServerInboundMessageTest.java
@@ -0,0 +1,72 @@
+package io.quarkus.grpc.server;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.grpc.ManagedChannel;
+import io.grpc.StatusRuntimeException;
+import io.grpc.examples.helloworld.GreeterGrpc;
+import io.grpc.examples.helloworld.HelloReply;
+import io.grpc.examples.helloworld.HelloRequest;
+import io.grpc.netty.NettyChannelBuilder;
+import io.quarkus.grpc.server.services.HelloService;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class GrpcServerInboundMessageTest {
+
+    static String configuration = """
+            quarkus.grpc.server.use-separate-server=false
+            quarkus.grpc.server.max-inbound-message-size=512000
+            """;
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest().setArchiveProducer(
+            () -> ShrinkWrap.create(JavaArchive.class)
+                    .addPackage(GreeterGrpc.class.getPackage())
+                    .addClass(HelloService.class)
+                    .add(new StringAsset(configuration), "application.properties"));
+
+    protected ManagedChannel channel;
+
+    @BeforeEach
+    public void init() throws Exception {
+        channel = NettyChannelBuilder.forAddress("localhost", 8081)
+                .usePlaintext()
+                .build();
+    }
+
+    @AfterEach
+    public void shutdown() {
+        if (channel != null) {
+            channel.shutdownNow();
+        }
+    }
+
+    @Test
+    public void testInvokingWithPayloadUnderLimit() {
+        var sizeInChars = 400 * 1024;
+        HelloRequest request = HelloRequest.newBuilder().setName("a".repeat(sizeInChars)).build();
+
+        HelloReply reply = GreeterGrpc.newBlockingStub(channel)
+                .sayHello(request);
+        assertThat(reply).isNotNull();
+    }
+
+    @Test
+    public void testInvokingWithPayloadAboveLimit() {
+        var sizeInChars = 1000 * 1024;
+        HelloRequest request = HelloRequest.newBuilder().setName("a".repeat(sizeInChars)).build();
+
+        assertThatThrownBy(() -> GreeterGrpc.newBlockingStub(channel).sayHello(request))
+                .isInstanceOf(StatusRuntimeException.class)
+                .hasMessageContaining("RESOURCE_EXHAUSTED");
+    }
+}
diff --git a/.../src/test/java/io/quarkus/grpc/server/GrpcServerInboundMessageWithSeparateServerTest.java b/.../src/test/java/io/quarkus/grpc/server/GrpcServerInboundMessageWithSeparateServerTest.java
@@ -0,0 +1,71 @@
+package io.quarkus.grpc.server;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.grpc.ManagedChannel;
+import io.grpc.StatusRuntimeException;
+import io.grpc.examples.helloworld.GreeterGrpc;
+import io.grpc.examples.helloworld.HelloReply;
+import io.grpc.examples.helloworld.HelloRequest;
+import io.grpc.netty.NettyChannelBuilder;
+import io.quarkus.grpc.server.services.HelloService;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class GrpcServerInboundMessageWithSeparateServerTest {
+
+    static String configuration = """
+            quarkus.grpc.server.max-inbound-message-size=512000
+            """;
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest().setArchiveProducer(
+            () -> ShrinkWrap.create(JavaArchive.class)
+                    .addPackage(GreeterGrpc.class.getPackage())
+                    .addClass(HelloService.class)
+                    .add(new StringAsset(configuration), "application.properties"));
+
+    protected ManagedChannel channel;
+
+    @BeforeEach
+    public void init() throws Exception {
+        channel = NettyChannelBuilder.forAddress("localhost", 9001)
+                .usePlaintext()
+                .build();
+    }
+
+    @AfterEach
+    public void shutdown() {
+        if (channel != null) {
+            channel.shutdownNow();
+        }
+    }
+
+    @Test
+    public void testInvokingWithPayloadUnderLimit() {
+        var sizeInChars = 400 * 1024;
+        HelloRequest request = HelloRequest.newBuilder().setName("a".repeat(sizeInChars)).build();
+
+        HelloReply reply = GreeterGrpc.newBlockingStub(channel)
+                .sayHello(request);
+        assertThat(reply).isNotNull();
+    }
+
+    @Test
+    public void testInvokingWithPayloadAboveLimit() {
+        var sizeInChars = 1000 * 1024;
+        HelloRequest request = HelloRequest.newBuilder().setName("a".repeat(sizeInChars)).build();
+
+        assertThatThrownBy(() -> GreeterGrpc.newBlockingStub(channel).sayHello(request))
+                .isInstanceOf(StatusRuntimeException.class)
+                .hasMessageContaining("RESOURCE_EXHAUSTED");
+    }
+}
diff --git a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java
@@ -76,6 +76,7 @@
 import io.vertx.grpc.VertxServer;
 import io.vertx.grpc.VertxServerBuilder;
 import io.vertx.grpc.server.GrpcServer;
+import io.vertx.grpc.server.GrpcServerOptions;
 import io.vertx.grpc.server.GrpcServiceBridge;
 
 @Recorder
@@ -146,7 +147,11 @@ private void buildGrpcServer(Vertx vertx, GrpcServerConfiguration configuration,
             Map<String, List<String>> virtualMethodsPerService,
             GrpcContainer grpcContainer, LaunchMode launchMode, boolean securityPresent) {
 
-        GrpcServer server = GrpcServer.server(vertx);
+        GrpcServerOptions options = new GrpcServerOptions();
+        if (!configuration.maxInboundMessageSize.isEmpty()) {
+            options.setMaxMessageSize(configuration.maxInboundMessageSize.getAsInt());
+        }
+        GrpcServer server = GrpcServer.server(vertx, options);
         List<ServerInterceptor> globalInterceptors = grpcContainer.getSortedGlobalInterceptors();
 
         if (launchMode == LaunchMode.DEVELOPMENT) {

diff --git a/...ns/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/config/GrpcServerConfiguration.java b/...ns/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/config/GrpcServerConfiguration.java
@@ -59,6 +59,9 @@ public class GrpcServerConfiguration {
 
     /**
      * The max inbound message size in bytes.
+     * <p>
+     * When using a single server (using {@code quarkus.grpc.server.use-separate-server=false}), the default value is 256KB.
+     * When using a separate server (using {@code quarkus.grpc.server.use-separate-server=true}), the default value is 4MB.
      */
     @ConfigItem
     public OptionalInt maxInboundMessageSize;