Make grpc executor configurable (#113)

Author: ghostdogpr

protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcConfig.scala

@@ -1,12 +1,16 @@
 package com.devsisters.shardcake
 
+import java.util.concurrent.Executor
+
 /**
  * The configuration for the gRPC client.
+ *
  * @param maxInboundMessageSize the maximum message size allowed to be received by the grpc client
+ * @param executor a custom executor to pass to grpc-java when creating gRPC clients and servers
  */
-case class GrpcConfig(maxInboundMessageSize: Int)
+case class GrpcConfig(maxInboundMessageSize: Int, executor: Option[Executor])
 
 object GrpcConfig {
   val default: GrpcConfig =
-    GrpcConfig(maxInboundMessageSize = 32 * 1024 * 1024)
+    GrpcConfig(maxInboundMessageSize = 32 * 1024 * 1024, None)
 }

protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcPods.scala

@@ -25,13 +25,23 @@ class GrpcPods(
           map.get(pod) match {
             case Some((channel, _)) => ZIO.succeed((channel, map))
             case None               =>
-              val channel: ZManagedChannel =
-                ZManagedChannel.apply(
-                  ManagedChannelBuilder
-                    .forAddress(pod.host, pod.port)
-                    .maxInboundMessageSize(config.maxInboundMessageSize)
-                    .usePlaintext()
-                )
+              val builder = {
+                config.executor match {
+                  case Some(executor) =>
+                    ManagedChannelBuilder
+                      .forAddress(pod.host, pod.port)
+                      .executor(executor)
+                      .maxInboundMessageSize(config.maxInboundMessageSize)
+                      .usePlaintext()
+                  case None           =>
+                    ManagedChannelBuilder
+                      .forAddress(pod.host, pod.port)
+                      .maxInboundMessageSize(config.maxInboundMessageSize)
+                      .usePlaintext()
+                }
+              }
+
+              val channel = ZManagedChannel(builder)
               // create a fiber that never ends and keeps the connection alive
               for {
                 _          <- ZIO.logDebug(s"Opening connection to pod $pod")

protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcShardingService.scala

@@ -53,14 +53,22 @@ object GrpcShardingService {
   /**
    * A layer that creates a gRPC server that exposes the Pods API.
    */
-  val live: ZLayer[Config with Sharding, Throwable, Unit] =
-    ZLayer.scoped[Config with Sharding] {
+  val live: ZLayer[Config with Sharding with GrpcConfig, Throwable, Unit] =
+    ZLayer.scoped[Config with Sharding with GrpcConfig] {
       for {
-        config   <- ZIO.service[Config]
-        sharding <- ZIO.service[Sharding]
-        builder   = ServerBuilder.forPort(config.shardingPort).addService(ProtoReflectionService.newInstance())
-        services  = ServiceList.add(new GrpcShardingService(sharding, config.sendTimeout) {})
-        _        <- ScopedServer.fromServiceList(builder, services)
+        config     <- ZIO.service[Config]
+        grpcConfig <- ZIO.service[GrpcConfig]
+        sharding   <- ZIO.service[Sharding]
+        builder     = grpcConfig.executor match {
+                        case Some(executor) =>
+                          ServerBuilder
+                            .forPort(config.shardingPort)
+                            .executor(executor)
+                        case None           =>
+                          ServerBuilder.forPort(config.shardingPort)
+                      }
+        services    = ServiceList.add(new GrpcShardingService(sharding, config.sendTimeout) {})
+        _          <- ScopedServer.fromServiceList(builder.addService(ProtoReflectionService.newInstance()), services)
       } yield ()
     }
 }