wip: update test to also cover completing promise with failure

Author: bpholt

twitter-futures/src/main/scala/com/dwolla/util/async/twitter.scala

@@ -3,11 +3,14 @@ package com.dwolla.util.async
 import cats.*
 import cats.data.*
 import cats.effect.*
+import cats.effect.std.Dispatcher
 import cats.syntax.all.*
+import cats.effect.syntax.all.*
 import cats.tagless.*
 import cats.tagless.syntax.all.*
 import com.dwolla.util.async.twitter.CancelledViaCatsEffect
 import com.twitter.util
+import com.twitter.util.{Future, Promise}
 
 import java.util.concurrent.CancellationException
 import scala.util.control.NoStackTrace
@@ -27,6 +30,20 @@ object twitter extends ToAsyncFunctorKOps {
   private[async] case object CancelledViaCatsEffect
     extends CancellationException("Cancelled via cats-effect")
       with NoStackTrace
+
+  class UnsafeFToFuture[F[_] : Sync](dispatcher: Dispatcher[F]) extends (F ~> Future) {
+    override def apply[A](fa: F[A]): Future[A] = {
+      val p = Promise[A]()
+      dispatcher.unsafeRunAndForget {
+        fa.attempt.flatMap {
+            case Left(ex) => Sync[F].delay(p.setException(ex))
+            case Right(a) => Sync[F].delay(p.setValue(a))
+          }
+          .onCancel(Sync[F].delay(p.raise(CancelledViaCatsEffect)))
+      }
+      p
+    }
+  }
 }
 
 class PartiallyAppliedProvide[F[_]](private val dummy: Boolean = true) extends AnyVal {
@@ -51,8 +68,10 @@ class PartiallyAppliedLiftFuture[F[_]] {
             .flatMap { fa =>
               Sync[F].delay {
                 fa.respond {
-                  case util.Return(a) => cb(Right(a))
-                  case util.Throw(ex) => cb(Left(ex))
+                  case util.Return(a) =>
+                    cb(Right(a))
+                  case util.Throw(ex) =>
+                    cb(Left(ex))
                 }
               }
             }

twitter-futures/src/test/scala/com/dwolla/util/async/twitter/TwitterFutureAsyncMapKTests.scala

@@ -1,16 +1,19 @@
 package com.dwolla.util.async
 
+import cats.*
 import cats.effect.*
 import cats.effect.std.*
 import cats.syntax.all.*
 import com.dwolla.util.async.twitter.{CancelledViaCatsEffect, liftFuture}
 import com.twitter.util.{Duration as _, *}
 import munit.{CatsEffectSuite, ScalaCheckEffectSuite}
-import org.scalacheck.{Prop, Test}
+import org.scalacheck.Arbitrary.arbitrary
 import org.scalacheck.effect.PropF
+import org.scalacheck.{Arbitrary, Gen, Prop, Test}
 
 import java.util.concurrent.CancellationException
 import scala.concurrent.duration.*
+import scala.util.control.NoStackTrace
 
 class TwitterFutureAsyncMapKTests extends CatsEffectSuite with ScalaCheckEffectSuite {
   override def munitIOTimeout: Duration = 1.minute
@@ -43,25 +46,34 @@ class TwitterFutureAsyncMapKTests extends CatsEffectSuite with ScalaCheckEffectS
     }
   }
 
-  test("a running Twitter Future lifted into IO can be completed or cancelled") {
-    PropF.forAllF { (i: Option[Int]) =>
+//  private val supervisorAndDispatcher = ResourceSuiteLocalFixture("supervisorAndDispatcher",
+//    (Supervisor[IO](await = true), Dispatcher.sequential[IO](await = true).onFinalize(IO.println("shutting down Dispatcher"))).tupled
+//  )
+
+//  override def munitFixtures: Seq[AnyFixture[?]] = super.munitFixtures ++ Seq(supervisorAndDispatcher)
+
+  test("a running Twitter Future lifted into IO can be completed (as success or failure) or cancelled") {
+    PropF.forAllF { (i: Outcome[IO, Throwable, Int]) =>
+      //      val (supervisor, dispatcher) = supervisorAndDispatcher()
+
       (Supervisor[IO](await = true), Dispatcher.parallel[IO](await = true))
         .tupled
         .use { case (supervisor, dispatcher) =>
           for {
+            expectedResult <- i.embed(CancelledViaCatsEffect.raiseError[IO, Int]).attempt
             capturedInterruptionThrowable <- Deferred[IO, Throwable]
-            twitterPromise <- IO(new Promise[Option[Int]]()).flatTap(captureThrowableOnInterruption(dispatcher, capturedInterruptionThrowable))
+            twitterPromise <- IO(new Promise[Int]()).flatTap(captureThrowableOnInterruption(dispatcher, capturedInterruptionThrowable))
             startedLatch <- CountDownLatch[IO](1)
             promiseFiber <- IO.uncancelable { poll => // we want only the Future to be cancellable
               supervisor.supervise(poll(liftFuture[IO](startedLatch.release.as(twitterPromise))))
             }
             _ <- startedLatch.await
             _ <- completeOrCancel(i, twitterPromise, promiseFiber)
-            cancelledRef <- Ref[IO].of(false)
-            outcome <- promiseFiber.joinWith(cancelledRef.set(true).as(None))
-            wasCancelled <- cancelledRef.get
+            outcome <- promiseFiber.join
+
+            outcomeEmittedValue <- outcome.embed(CancelledViaCatsEffect.raiseError[IO, Int]).attempt
 
-            expectCancellation = i.isEmpty
+            expectCancellation = i.isCanceled
             _ <- interceptMessageIO[CancellationException]("Cancelled via cats-effect") {
               capturedInterruptionThrowable
                 .get
@@ -71,8 +83,8 @@ class TwitterFutureAsyncMapKTests extends CatsEffectSuite with ScalaCheckEffectS
             }
               .whenA(expectCancellation)
           } yield {
-            assertEquals(outcome, i)
-            assertEquals(wasCancelled, i.as(false).getOrElse(true))
+            assertEquals(outcomeEmittedValue, expectedResult)
+            assertEquals(outcome.isCanceled, i.isCanceled)
             assertEquals(Option(CancelledViaCatsEffect).filter(_ => expectCancellation), twitterPromise.isInterrupted)
           }
         }
@@ -90,6 +102,14 @@ class TwitterFutureAsyncMapKTests extends CatsEffectSuite with ScalaCheckEffectS
     }
   }
 
+  private def genOutcome[F[_] : Applicative, A: Arbitrary]: Gen[Outcome[F, Throwable, A]] =
+    Gen.oneOf(
+      arbitrary[A].map(_.pure[F]).map(Outcome.succeeded[F, Throwable, A]),
+      Gen.const(new RuntimeException("arbitrary exception") with NoStackTrace).map(Outcome.errored[F, Throwable, A]),
+      Gen.const(Outcome.canceled[F, Throwable, A]),
+    )
+  private implicit def arbOutcome[F[_] : Applicative, A: Arbitrary]: Arbitrary[Outcome[F, Throwable, A]] = Arbitrary(genOutcome)
+
   private def captureThrowableOnInterruption[F[_] : Sync, A](dispatcher: Dispatcher[F],
                                                              capture: Deferred[F, Throwable])
                                                             (p: Promise[A]): F[Unit] =
@@ -99,14 +119,17 @@ class TwitterFutureAsyncMapKTests extends CatsEffectSuite with ScalaCheckEffectS
       }
     }
 
-  private def completeOrCancel[F[_] : Sync, A](maybeA: Option[A],
-                                               promise: Promise[Option[A]],
-                                               fiber: Fiber[F, Throwable, Option[A]]): F[Unit] =
+  private def completeOrCancel[F[_] : Sync, A](maybeA: Outcome[F, Throwable, A],
+                                               promise: Promise[A],
+                                               fiber: Fiber[F, Throwable, A]): F[Unit] =
     maybeA match {
-      case Some(a) => Sync[F].delay {
-        promise.setValue(a.some)
-      }
-      case None =>
+      case Outcome.Succeeded(fa) =>
+        fa.flatMap(a => Sync[F].delay(promise.setValue(a))).void
+
+      case Outcome.Errored(ex) =>
+        Sync[F].delay(promise.setException(ex))
+
+      case Outcome.Canceled() =>
         fiber.cancel
     }