Merge pull request #717 from fredfp/remove_fix_blocking

Fix: block in F using F.blocking instead of scala.concurrent.blocking
typelevel · Aug 21, 2024 · 30a1d11 · 30a1d11
2 parents 35c67b2 + 1d9fa2c
commit 30a1d11
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 16 deletions.
diff --git a/runtime/src/main/scala/fs2/grpc/syntax/ManagedChannelBuilderSyntax.scala b/runtime/src/main/scala/fs2/grpc/syntax/ManagedChannelBuilderSyntax.scala
@@ -24,8 +24,8 @@ package grpc
 package syntax
 
 import java.util.concurrent.TimeUnit
-import scala.concurrent._
 import cats.effect._
+import cats.syntax.all._
 import io.grpc.{ManagedChannel, ManagedChannelBuilder}
 
 trait ManagedChannelBuilderSyntax {
@@ -48,13 +48,11 @@ final class ManagedChannelBuilderOps[MCB <: ManagedChannelBuilder[MCB]](val buil
     */
   def resource[F[_]](implicit F: Sync[F]): Resource[F, ManagedChannel] =
     resourceWithShutdown { ch =>
-      F.delay {
-        ch.shutdown()
-        if (!blocking(ch.awaitTermination(30, TimeUnit.SECONDS))) {
-          ch.shutdownNow()
-          ()
-        }
-      }
+      for {
+        _ <- F.delay(ch.shutdown())
+        terminated <- F.interruptible(ch.awaitTermination(30, TimeUnit.SECONDS))
+        _ <- F.unlessA(terminated)(F.delay(ch.shutdownNow()))
+      } yield (())
     }
 
   /** Builds a `ManagedChannel` into a resource. The managed channel is shut down when the resource is released.

diff --git a/runtime/src/main/scala/fs2/grpc/syntax/ServerBuilderSyntax.scala b/runtime/src/main/scala/fs2/grpc/syntax/ServerBuilderSyntax.scala
@@ -24,8 +24,8 @@ package grpc
 package syntax
 
 import java.util.concurrent.TimeUnit
-import scala.concurrent._
 import cats.effect._
+import cats.syntax.all._
 import io.grpc.{Server, ServerBuilder}
 
 trait ServerBuilderSyntax {
@@ -45,13 +45,11 @@ final class ServerBuilderOps[SB <: ServerBuilder[SB]](val builder: SB) extends A
     */
   def resource[F[_]](implicit F: Sync[F]): Resource[F, Server] =
     resourceWithShutdown { server =>
-      F.delay {
-        server.shutdown()
-        if (!blocking(server.awaitTermination(30, TimeUnit.SECONDS))) {
-          server.shutdownNow()
-          ()
-        }
-      }
+      for {
+        _ <- F.delay(server.shutdown())
+        terminated <- F.interruptible(server.awaitTermination(30, TimeUnit.SECONDS))
+        _ <- F.unlessA(terminated)(F.delay(server.shutdownNow()))
+      } yield (())
     }
 
   /** Builds a `Server` into a resource. The server is shut down when the resource is released.