feat(channel-registry): expose channel config (#23)

Author: skjindal93

grpc-client-rx-utils/build.gradle.kts

@@ -7,15 +7,21 @@ plugins {
 
 dependencies {
   api("io.reactivex.rxjava3:rxjava:3.0.6")
-  api("io.grpc:grpc-stub:1.40.0")
+  api("io.grpc:grpc-stub:1.42.0")
   api(project(":grpc-context-utils"))
-  implementation("io.grpc:grpc-context:1.40.0")
+  implementation("io.grpc:grpc-context:1.42.0")
 
+  constraints {
+    implementation("com.google.code.gson:gson:2.8.9") {
+      because("https://snyk.io/vuln/SNYK-JAVA-COMGOOGLECODEGSON-1730327")
+    }
+  }
+  
   testImplementation("org.junit.jupiter:junit-jupiter:5.7.0")
   testImplementation("org.mockito:mockito-core:3.12.1")
   testImplementation("org.mockito:mockito-junit-jupiter:3.12.1")
 }
 
 tasks.test {
   useJUnitPlatform()
-}
+}

grpc-client-utils/build.gradle.kts

@@ -6,16 +6,25 @@ plugins {
 }
 
 dependencies {
-  api("io.grpc:grpc-context:1.40.0")
-  api("io.grpc:grpc-api:1.40.0")
+  api("io.grpc:grpc-context:1.42.0")
+  api("io.grpc:grpc-api:1.42.0")
 
   implementation(project(":grpc-context-utils"))
   implementation("org.slf4j:slf4j-api:1.7.30")
 
+  annotationProcessor("org.projectlombok:lombok:1.18.18")
+  compileOnly("org.projectlombok:lombok:1.18.18")
+
+  constraints {
+    implementation("com.google.code.gson:gson:2.8.9") {
+      because("https://snyk.io/vuln/SNYK-JAVA-COMGOOGLECODEGSON-1730327")
+    }
+  }
+
   testImplementation("org.junit.jupiter:junit-jupiter:5.7.0")
   testImplementation("org.mockito:mockito-core:3.12.1")
   testImplementation("org.mockito:mockito-inline:3.12.1")
-  testRuntimeOnly("io.grpc:grpc-netty:1.40.0")
+  testRuntimeOnly("io.grpc:grpc-netty:1.42.0")
 }
 
 tasks.test {

grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/GrpcChannelConfig.java

@@ -0,0 +1,10 @@
+package org.hypertrace.core.grpcutils.client;
+
+import lombok.Builder;
+import lombok.Value;
+
+@Value
+@Builder
+public class GrpcChannelConfig {
+  Integer maxInboundMessageSize;
+}

grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/GrpcChannelRegistry.java

@@ -6,6 +6,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import org.slf4j.Logger;
@@ -21,36 +22,50 @@ public class GrpcChannelRegistry {
    */
   @Deprecated
   public ManagedChannel forAddress(String host, int port) {
-    return this.forPlaintextAddress(host, port);
+    return this.forPlaintextAddress(host, port, GrpcChannelConfig.builder().build());
   }
 
   public ManagedChannel forSecureAddress(String host, int port) {
+    return forSecureAddress(host, port, GrpcChannelConfig.builder().build());
+  }
+
+  public ManagedChannel forSecureAddress(String host, int port, GrpcChannelConfig config) {
     assert !this.isShutdown;
-    String channelId = this.getChannelId(host, port, false);
+    String channelId = this.getChannelId(host, port, false, config);
     return this.channelMap.computeIfAbsent(
-        channelId, unused -> this.buildNewChannel(host, port, false));
+        channelId, unused -> this.buildNewChannel(host, port, false, config));
   }
 
   public ManagedChannel forPlaintextAddress(String host, int port) {
+    return forPlaintextAddress(host, port, GrpcChannelConfig.builder().build());
+  }
+
+  public ManagedChannel forPlaintextAddress(String host, int port, GrpcChannelConfig config) {
     assert !this.isShutdown;
-    String channelId = this.getChannelId(host, port, true);
+    String channelId = this.getChannelId(host, port, true, config);
     return this.channelMap.computeIfAbsent(
-        channelId, unused -> this.buildNewChannel(host, port, true));
+        channelId, unused -> this.buildNewChannel(host, port, true, config));
   }
 
-  private ManagedChannel buildNewChannel(String host, int port, boolean isPlaintext) {
-    LOG.info("Creating new channel {}", this.getChannelId(host, port, isPlaintext));
+  private ManagedChannel buildNewChannel(
+      String host, int port, boolean isPlaintext, GrpcChannelConfig config) {
+    LOG.info("Creating new channel {}", this.getChannelId(host, port, isPlaintext, config));
 
     ManagedChannelBuilder<?> builder = ManagedChannelBuilder.forAddress(host, port);
     if (isPlaintext) {
       builder.usePlaintext();
     }
+
+    if (config.getMaxInboundMessageSize() != null) {
+      builder.maxInboundMessageSize(config.getMaxInboundMessageSize());
+    }
     return builder.build();
   }
 
-  private String getChannelId(String host, int port, boolean isPlaintext) {
+  private String getChannelId(
+      String host, int port, boolean isPlaintext, GrpcChannelConfig config) {
     String securePrefix = isPlaintext ? "plaintext" : "secure";
-    return securePrefix + ":" + host + ":" + port;
+    return securePrefix + ":" + host + ":" + port + ":" + Objects.hash(config);
   }
 
   /**

grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/GrpcChannelRegistryTest.java

@@ -50,6 +50,23 @@ void reusesChannelsForDuplicateRequests() {
     assertNotSame(firstChannelSecure, this.channelRegistry.forSecureAddress("bar", 1000));
   }
 
+  @Test
+  void setsMaxInboundMessageSizeConfig() {
+    Channel channel =
+        this.channelRegistry.forPlaintextAddress(
+            "foo", 1000, GrpcChannelConfig.builder().maxInboundMessageSize(100).build());
+    // same message size
+    assertSame(
+        channel,
+        this.channelRegistry.forPlaintextAddress(
+            "foo", 1000, GrpcChannelConfig.builder().maxInboundMessageSize(100).build()));
+    // different message size
+    assertNotSame(
+        channel,
+        this.channelRegistry.forPlaintextAddress(
+            "foo", 1000, GrpcChannelConfig.builder().maxInboundMessageSize(200).build()));
+  }
+
   @SuppressWarnings("rawtypes")
   @Test
   void shutdownAllChannelsOnShutdown() throws InterruptedException {

grpc-context-utils/build.gradle.kts

@@ -11,7 +11,7 @@ tasks.test {
 
 dependencies {
   // grpc
-  implementation("io.grpc:grpc-core:1.40.0")
+  implementation("io.grpc:grpc-core:1.42.0")
 
   implementation("com.auth0:java-jwt:3.14.0")
   implementation("com.auth0:jwks-rsa:0.17.0")
@@ -21,6 +21,12 @@ dependencies {
   implementation("org.slf4j:slf4j-api:1.7.30")
   // End Logging
 
+  constraints {
+    implementation("com.google.code.gson:gson:2.8.9") {
+      because("https://snyk.io/vuln/SNYK-JAVA-COMGOOGLECODEGSON-1730327")
+    }
+  }
+
   testImplementation("org.junit.jupiter:junit-jupiter:5.7.0")
   testImplementation("org.mockito:mockito-core:3.12.1")
 }

grpc-server-rx-utils/build.gradle.kts

@@ -7,7 +7,7 @@ plugins {
 
 dependencies {
   api("io.reactivex.rxjava3:rxjava:3.0.6")
-  api("io.grpc:grpc-stub:1.40.0")
+  api("io.grpc:grpc-stub:1.42.0")
 
   annotationProcessor("org.projectlombok:lombok:1.18.18")
   compileOnly("org.projectlombok:lombok:1.18.18")
@@ -18,6 +18,10 @@ dependencies {
     implementation("com.google.guava:guava:30.1-jre") {
       because("https://snyk.io/vuln/SNYK-JAVA-COMGOOGLEGUAVA-1015415")
     }
+    implementation("com.google.code.gson:gson:2.8.9") {
+      because("https://snyk.io/vuln/SNYK-JAVA-COMGOOGLECODEGSON-1730327")
+    }
+
   }
   testImplementation("org.junit.jupiter:junit-jupiter:5.7.0")
   testImplementation("org.mockito:mockito-core:3.12.1")

grpc-server-utils/build.gradle.kts

@@ -10,8 +10,8 @@ tasks.test {
 }
 
 dependencies {
-  api("io.grpc:grpc-context:1.40.0")
-  api("io.grpc:grpc-api:1.40.0")
+  api("io.grpc:grpc-context:1.42.0")
+  api("io.grpc:grpc-api:1.42.0")
 
   implementation(project(":grpc-context-utils"))
   implementation("org.slf4j:slf4j-api:1.7.30")