Scala 635 cancellation in cats effect (#1194)

* SCALA-635: add examples on Cancellation in Cats Effect

* SCALA-635: bump cats-effect to 3.5.3

Author: SergeiIonin

cats-effects/src/main/scala/com/baeldung/scala/catseffects/Cancellation.scala

@@ -0,0 +1,83 @@
+package com.baeldung.scala.catseffects
+
+import cats.effect.{IO, Resource}
+import cats.implicits.{catsSyntaxParallelTraverse1, toTraverseOps}
+
+import java.io.{InputStream, OutputStream}
+import java.util.concurrent.atomic.AtomicBoolean
+import scala.concurrent.duration.DurationInt
+
+object Cancellation {
+
+  def cancelFiberDirectly[A](io: IO[A], onCancellation: => IO[Unit]): IO[Unit] =
+    for {
+      fiber <- io.onCancel(onCancellation).start
+      _ <- fiber.cancel
+      _ <- fiber.join
+    } yield ()
+
+  def naiveParMap_1[A, B, C](
+    ioa: IO[A],
+    iob: IO[B],
+    onCancelA: IO[A],
+    onCancelB: IO[B],
+    f: (A, B) => C
+  ): IO[C] =
+    for {
+      fiberA <- ioa.start
+      fiberB <- iob.start
+      a <- fiberA.joinWith(onCancelA).onError(_ => fiberB.cancel)
+      b <- fiberB.joinWith(onCancelB).onError(_ => fiberA.cancel)
+    } yield f(a, b)
+
+  def naiveParMap_2[A, B, C](
+    ioa: IO[A],
+    iob: IO[B],
+    onCancelA: IO[A],
+    onCancelB: IO[B],
+    f: (A, B) => C
+  ): IO[C] =
+    for {
+      fiberA <- ioa.start
+      fiberB <- iob.start
+      fiberAj = fiberA.joinWith(onCancelA).onError(_ => fiberB.cancel)
+      fiberBj = fiberB.joinWith(onCancelB).onError(_ => fiberA.cancel)
+      regA <- fiberAj.start
+      regB <- fiberBj.start
+      a <- regA.joinWith(onCancelA)
+      b <- regB.joinWith(onCancelB)
+    } yield f(a, b)
+
+  def getEmails(): IO[List[String]] = IO(
+    List("[email protected]", "[email protected]", "[email protected]")
+  )
+
+  def send(message: String)(email: String): IO[Unit] = IO.sleep(
+    1.seconds
+  ) >> IO.println(s"sending email to $email with $message")
+
+  def sendEmailsUncancelable(message: String): IO[Unit] =
+    for {
+      emails <- getEmails()
+      _ <- IO.println("ready to send emails")
+      _ <- IO.uncancelable(_ => emails.traverse(send(message)))
+      _ <- IO.println("emails are sent")
+    } yield ()
+
+  def sendEmailsUncancelableResource(message: String) =
+    Resource
+      .make(getEmails())(emails => IO.println(s"emails $emails are released"))
+      .use(emails =>
+        IO.println("ready to send emails") >>
+          IO.uncancelable(_ => emails.traverse(send(message))) >>
+          IO.println("emails are sent")
+      )
+
+  def sendEmailsUncancelableBracket(message: String): IO[Unit] =
+    (getEmails()).bracket(emails =>
+      IO.println("ready to send emails") >>
+        IO.uncancelable(_ => emails.traverse(send(message))) >>
+        IO.println("emails are sent")
+    )(emails => IO.println(s"[bracket] emails $emails are released"))
+
+}

cats-effects/src/main/scala/com/baeldung/scala/catseffects/CancellationApp.scala

@@ -0,0 +1,31 @@
+package com.baeldung.scala.catseffects
+
+import cats.effect.IOApp
+import cats.effect.IO
+import scala.concurrent.duration.DurationInt
+import java.util.concurrent.atomic.AtomicBoolean
+import cats.effect.ExitCode
+
+object CancellationApp extends IOApp {
+
+  def example() = {
+    val flag = new AtomicBoolean(false)
+    var counter = 0
+    val ioa = IO.blocking {
+      while (!flag.get()) {
+        Thread.sleep(500)
+        println(s"counter = $counter")
+        counter += 1
+      }
+    }
+
+    ioa.cancelable(
+      IO.sleep(3.seconds) >> IO.println("executing the finalizer...") >> IO
+        .delay(flag.set(true))
+    )
+  }
+
+  override def run(args: List[String]) =
+    example().as(ExitCode.Success)
+
+}

cats-effects/src/test/scala/com/baeldung/scala/catseffects/CancellationSpec.scala

@@ -0,0 +1,106 @@
+package com.baeldung.scala.catseffects
+
+import cats.effect.kernel.CancelScope.Cancelable
+import cats.effect.{IO, Resource}
+import cats.effect.unsafe.implicits.global
+import cats.implicits.catsSyntaxTuple2Parallel
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.concurrent.duration.DurationInt
+
+class CancellationSpec extends AnyWordSpec with Matchers {
+
+  "Cancellation" should {
+
+    "cancel the fiber directly and execute the action on cancellation" in {
+      val flag = new AtomicBoolean(false)
+      val io = IO.println("ready to cancel the fiber") >> IO.sleep(
+        5.seconds
+      ) >> IO.println("Hello, World!")
+      val onCancel =
+        IO.println("cancellation signal received") >> IO.delay(flag.set(true))
+      val cancelFiber = Cancellation.cancelFiberDirectly(io, onCancel)
+      cancelFiber.unsafeRunSync()
+      flag.get() shouldBe true
+    }
+
+    "naive parMap works" in {
+      def tickingClock: IO[Unit] =
+        for {
+          _ <- IO.println(s"current time = ${System.currentTimeMillis()}")
+          _ <- IO.sleep(250.millisecond)
+          _ <- tickingClock
+        } yield ()
+      val error: IO[Unit] =
+        IO.sleep(1.second) *> IO.raiseError(new RuntimeException("boom!"))
+
+      val parMapNaive_2 = Cancellation.naiveParMap_2(
+        tickingClock,
+        error,
+        IO.println("[2] tickingClock was cancelled"),
+        IO.println("[2] error was cancelled"),
+        (_: Unit, _: Unit) => println("[2] Exit")
+      )
+
+      val res = parMapNaive_2.handleError(e => e.getMessage).unsafeRunSync()
+      res shouldBe "boom!"
+    }
+
+    "send emails if the cancellation signal came during the sending process" in {
+      val sendEmails: IO[Unit] =
+        Cancellation
+          .sendEmailsUncancelable("Exam on FP is shifted to Jan 11, 2024")
+          .onCancel(IO.println("cancellation signal received"))
+      IO.race(IO.sleep(1.seconds), sendEmails).unsafeRunSync()
+    }
+
+    "if one of the effects running in parallel throws an error, the second one in cancelled, but it won't affect uncancelalble regions" in {
+      val flag = new AtomicBoolean(false)
+      val res = (
+        IO.sleep(500.millisecond) >> IO.raiseError(
+          new RuntimeException("Boom!")
+        ),
+        IO.uncancelable(_ =>
+          IO.sleep(1.second) >> IO.println("Hey there") >> IO.delay(
+            flag.set(true)
+          )
+        )
+      ).parTupled
+        .handleError(_ => flag.get)
+        .unsafeRunSync()
+      res shouldBe true
+    }
+
+    "partially cancel the unancelable IO" in {
+      var counter = 0
+      val example: IO[Unit] = {
+        val flag = new AtomicBoolean(false)
+        val ioa = IO.blocking {
+          while (!flag.get()) {
+            Thread.sleep(100)
+            println(s"counter = $counter")
+            counter += 1
+          }
+        }
+
+        ioa.cancelable(
+          IO.println("executing the finalizer...") >> IO.delay(flag.set(true))
+        )
+      }
+
+      (for {
+        fiber <- example.start
+        _ <- IO.sleep(500.millisecond)
+        _ <- IO.println("cancelling the fiber")
+        _ <- fiber.cancel
+        _ <- fiber.join
+      } yield ()).unsafeRunSync()
+
+      counter <= 6 shouldBe true
+    }
+
+  }
+
+}