Merge pull request #2907 from armanbilge/issue/2903

Fix `Async#fromCompletableFuture` cancelation
typelevel · Mar 24, 2022 · 8956adf · 8956adf
2 parents 8d8f074 + 4ec557d
commit 8956adf
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 1 deletion.
diff --git a/kernel/jvm/src/main/scala/cats/effect/kernel/AsyncPlatform.scala b/kernel/jvm/src/main/scala/cats/effect/kernel/AsyncPlatform.scala
@@ -40,7 +40,14 @@ private[kernel] trait AsyncPlatform[F[_]] { this: Async[F] =>
               }))
           }
 
-          Some(void(delay(cf.cancel(false))))
+          Some(
+            ifM(delay(cf.cancel(false)))(
+              unit,
+              async_[Unit] { cb =>
+                cf.handle[Unit]((_, _) => cb(Right(())))
+                ()
+              }
+            ))
         }
       }
     }

diff --git a/tests/jvm/src/test/scala/cats/effect/kernel/AsyncPlatformSpec.scala b/tests/jvm/src/test/scala/cats/effect/kernel/AsyncPlatformSpec.scala
@@ -40,5 +40,20 @@ class AsyncPlatformSpec extends BaseSpec {
         _ <- IO(cf.join() must throwA[CancellationException])
       } yield ok
     }
+
+    "backpressure on CompletableFuture cancelation" in ticked { implicit ticker =>
+      // a non-cancelable, never-completing CompletableFuture
+      def cf = new CompletableFuture[Unit] {
+        override def cancel(mayInterruptIfRunning: Boolean) = false
+      }
+
+      val io = for {
+        fiber <- IO.fromCompletableFuture(IO(cf)).start
+        _ <- smallDelay // time for the callback to be set-up
+        _ <- fiber.cancel
+      } yield ()
+
+      io must nonTerminate
+    }
   }
 }