Specify executor for GRPC server/channel

core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectRouteBuilder.scala

@@ -10,6 +10,8 @@ import org.ivovk.connect_rpc_scala.http.*
 import org.ivovk.connect_rpc_scala.http.QueryParams.*
 import scalapb.json4s.{JsonFormat, Printer}
 
+import java.util.concurrent.Executor
+import scala.concurrent.ExecutionContext
 import scala.concurrent.duration.*
 
 object ConnectRouteBuilder {
@@ -28,8 +30,9 @@ object ConnectRouteBuilder {
 case class ConnectRouteBuilder[F[_] : Async] private(
   services: Seq[ServerServiceDefinition],
   jsonPrinterConfigurator: Endo[Printer] = identity,
-  serverBuilderConfigurator: Endo[ServerBuilder[_]] = identity,
-  channelBuilderConfigurator: Endo[ManagedChannelBuilder[_]] = identity,
+  serverConfigurator: Endo[ServerBuilder[_]] = identity,
+  channelConfigurator: Endo[ManagedChannelBuilder[_]] = identity,
+  executor: Executor = ExecutionContext.global,
   waitForShutdown: Duration = 5.seconds,
 ) {
 
@@ -38,11 +41,14 @@ case class ConnectRouteBuilder[F[_] : Async] private(
   def withJsonPrinterConfigurator(method: Endo[Printer]): ConnectRouteBuilder[F] =
     copy(jsonPrinterConfigurator = method)
 
-  def withServerBuilderConfigurator(method: Endo[ServerBuilder[_]]): ConnectRouteBuilder[F] =
-    copy(serverBuilderConfigurator = method)
+  def withServerConfigurator(method: Endo[ServerBuilder[_]]): ConnectRouteBuilder[F] =
+    copy(serverConfigurator = method)
 
-  def withChannelBuilderConfigurator(method: Endo[ManagedChannelBuilder[_]]): ConnectRouteBuilder[F] =
-    copy(channelBuilderConfigurator = method)
+  def withChannelConfigurator(method: Endo[ManagedChannelBuilder[_]]): ConnectRouteBuilder[F] =
+    copy(channelConfigurator = method)
+
+  def withExecutor(executor: Executor): ConnectRouteBuilder[F] =
+    copy(executor = executor)
 
   def withWaitForShutdown(duration: Duration): ConnectRouteBuilder[F] =
     copy(waitForShutdown = duration)
@@ -68,8 +74,9 @@ case class ConnectRouteBuilder[F[_] : Async] private(
     for
       channel <- InProcessChannelBridge.create(
         services,
-        serverBuilderConfigurator,
-        channelBuilderConfigurator,
+        serverConfigurator,
+        channelConfigurator,
+        executor,
         waitForShutdown,
       )
     yield

core/src/main/scala/org/ivovk/connect_rpc_scala/InProcessChannelBridge.scala

@@ -6,7 +6,7 @@ import io.grpc.*
 import io.grpc.inprocess.{InProcessChannelBuilder, InProcessServerBuilder}
 import org.slf4j.{Logger, LoggerFactory}
 
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{Executor, TimeUnit}
 import scala.concurrent.duration.Duration
 import scala.jdk.CollectionConverters.*
 import scala.util.chaining.*
@@ -17,36 +17,37 @@ object InProcessChannelBridge {
 
   def create[F[_] : Sync](
     services: Seq[ServerServiceDefinition],
-    serverBuilderConfigurator: Endo[ServerBuilder[?]] = identity,
-    channelBuilderConfigurator: Endo[ManagedChannelBuilder[?]] = identity,
+    serverConfigurator: Endo[ServerBuilder[?]],
+    channelConfigurator: Endo[ManagedChannelBuilder[?]],
+    executor: Executor,
     waitForShutdown: Duration,
   ): Resource[F, Channel] = {
     for
       name <- Resource.eval(Sync[F].delay(InProcessServerBuilder.generateName()))
-      server <- createServer(name, services, waitForShutdown, serverBuilderConfigurator)
-      channel <- createStub(name, waitForShutdown, channelBuilderConfigurator)
+      server <- createServer(name, services, serverConfigurator, executor, waitForShutdown)
+      channel <- createStub(name, channelConfigurator, executor, waitForShutdown)
     yield channel
   }
 
   private def createServer[F[_] : Sync](
     name: String,
     services: Seq[ServerServiceDefinition],
+    serverConfigurator: Endo[ServerBuilder[?]],
+    executor: Executor,
     waitForShutdown: Duration,
-    serverBuilderConfigurator: Endo[ServerBuilder[?]] = identity,
   ): Resource[F, Server] = {
     val acquire = Sync[F].delay {
       InProcessServerBuilder.forName(name)
-        .directExecutor()
         .addServices(services.asJava)
-        .pipe(serverBuilderConfigurator)
-        .build()
-        .start()
+        .executor(executor)
+        .pipe(serverConfigurator)
+        .build().start()
     }
     val release = (s: Server) => Sync[F].delay {
       val res = s.shutdown().awaitTermination(waitForShutdown.toMillis, TimeUnit.MILLISECONDS)
 
       if (!res) {
-        logger.warn(s"GRPC server did not shut down in $waitForShutdown")
+        logger.warn(s"GRPC server did not shut down in $waitForShutdown, consider increasing shutdown timeout")
       }
     }
 
@@ -55,20 +56,21 @@ object InProcessChannelBridge {
 
   private def createStub[F[_] : Sync](
     name: String,
+    channelConfigurator: Endo[ManagedChannelBuilder[?]],
+    executor: Executor,
     waitForShutdown: Duration,
-    channelBuilderConfigurator: Endo[ManagedChannelBuilder[?]] = identity,
   ): Resource[F, ManagedChannel] = {
     val acquire = Sync[F].delay {
       InProcessChannelBuilder.forName(name)
-        .directExecutor()
-        .pipe(channelBuilderConfigurator)
+        .executor(executor)
+        .pipe(channelConfigurator)
         .build()
     }
     val release = (c: ManagedChannel) => Sync[F].delay {
       val res = c.shutdown().awaitTermination(waitForShutdown.toMillis, TimeUnit.MILLISECONDS)
 
       if (!res) {
-        logger.warn(s"GRPC channel did not shut down in $waitForShutdown")
+        logger.warn(s"GRPC channel did not shut down in $waitForShutdown, consider increasing shutdown timeout")
       }
     }