As Reactormonk says in a comment above, it's not possible to write an instance of Async for Future that has the right semantics, because Async extends Sync, and Sync requires a representation of a computation that can be run repeatedly, while Scala's futures begin running when they're defined and can't be re-run.
An unlawful instance
It's instructive to see this for yourself, though, and I'd encourage you to try to write your own compile-able but (necessarily) unlawful Async[Future] instance without looking at the next block of code. For the sake of the example, though, here's a quick sketch off the top of my head:
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Failure, Success}
import cats.effect.{Async, ExitCase, IO}
def futureAsync(implicit c: ExecutionContext): Async[Future] = new Async[Future] {
def async[A](k: (Either[Throwable, A] => Unit) => Unit): Future[A] =
IO.async(k).unsafeToFuture()
def asyncF[A](k: (Either[Throwable, A] => Unit) => Future[Unit]): Future[A] = {
val p = Promise[A]()
val f = k {
case Right(a) => p.success(a)
case Left(e) => p.failure(e)
}
f.flatMap(_ => p.future)
}
def suspend[A](thunk: => Future[A]): Future[A] = Future(thunk).flatten
def bracketCase[A, B](acquire: Future[A])(use: A => Future[B])(
release: (A, ExitCase[Throwable]) => Future[Unit]
): Future[B] = acquire.flatMap { a =>
use(a).transformWith {
case Success(b) => release(a, ExitCase.Completed).map(_ => b)
case Failure(e) => release(a, ExitCase.Error(e)).flatMap(_ => Future.failed(e))
}
}
def raiseError[A](e: Throwable): Future[A] = Future.failed(e)
def handleErrorWith[A](fa: Future[A])(f: Throwable => Future[A]): Future[A] =
fa.recoverWith { case t => f(t) }
def pure[A](x: A): Future[A] = Future.successful(x)
def flatMap[A, B](fa: Future[A])(f: A => Future[B]): Future[B] = fa.flatMap(f)
def tailRecM[A, B](a: A)(f: A => Future[Either[A, B]]): Future[B] = f(a).flatMap {
case Right(b) => Future.successful(b)
case Left(a) => tailRecM(a)(f)
}
}
This will compile just fine, and would probably work for some situations (but please don't actually use it!). We've said it can't have the right semantics, though, and we can show that by using cats-effect's laws module.
Checking the laws
First we need some boilerplate-y stuff you don't really need to worry about:
import cats.kernel.Eq, cats.implicits._
import org.scalacheck.Arbitrary
implicit val throwableEq: Eq[Throwable] = Eq.by[Throwable, String](_.toString)
implicit val nonFatalArbitrary: Arbitrary[Throwable] =
Arbitrary(Arbitrary.arbitrary[Exception].map(identity))
implicit def futureEq[A](implicit A: Eq[A], ec: ExecutionContext): Eq[Future[A]] =
new Eq[Future[A]] {
private def liftToEither(f: Future[A]): Future[Either[Throwable, A]] =
f.map(Right(_)).recover { case e => Left(e) }
def eqv(fx: Future[A], fy: Future[A]): Boolean =
scala.concurrent.Await.result(
liftToEither(fx).zip(liftToEither(fy)).map {
case (rx, ry) => rx === ry
},
scala.concurrent.duration.Duration(1, "second")
)
}
Then we can define a test that checks the Async laws for our instance:
import cats.effect.laws.discipline.{AsyncTests, Parameters}
import org.scalatest.FunSuite
import org.typelevel.discipline.scalatest.Discipline
object FutureAsyncSuite extends FunSuite with Discipline {
implicit val ec: ExecutionContext = ExecutionContext.global
implicit val params: Parameters =
Parameters.default.copy(allowNonTerminationLaws = false)
checkAll(
"Async",
AsyncTests[Future](futureAsync).async[String, String, String]
)
}
And then we can run the law tests:
scala> FutureAsyncSuite.execute()
FutureAsyncSuite:
- Async.async.acquire and release of bracket are uncancelable
- Async.async.ap consistent with product + map
- Async.async.applicative homomorphism
...
You'll see that most of the tests are green; this instance gets a lot of things right.
Where it breaks the law
It does show three failed tests, though, including the following:
- Async.async.repeated sync evaluation not memoized *** FAILED ***
GeneratorDrivenPropertyCheckFailedException was thrown during property evaluation.
(Discipline.scala:14)
Falsified after 1 successful property evaluations.
Location: (Discipline.scala:14)
Occurred when passed generated values (
arg0 = "淳칇멀",
arg1 = org.scalacheck.GenArities$$Lambda$7154/1834868832@1624ea25
)
Label of failing property:
Expected: Future(Success(驅ṇ숆㽝珅뢈矉))
Received: Future(Success(淳칇멀))
If you look at the laws definitions, you'll see that this is a test that defines a Future value with delay and then sequences it multiple times, like this:
val change = F.delay { /* observable side effect here */ }
val read = F.delay(cur)
change *> change *> read
The other two failures are similar "not memoized" violations. These tests should see the side effect happen twice, but in our case it's not possible to write delay or suspend for Future in such a way that that would happen (it's worth trying, though, to convince yourself that this is the case).
What you should do instead
To sum up: you can write an Async[Future] instance that will pass something like 75 of the 78 Async laws tests, but it's not possible to write an instance that will pass all of them, and using an unlawful instance is a really bad idea: both potential users of your code and libraries like Doobie will assume that your instances are lawful, and if you don't live up to this assumption you're opening the door to complex and annoying bugs.
It's worth noting that it's not too hard to write a minimal wrapper for Future that has a lawful Async instance (for example I've got a wrapper for Twitter's future called Rerunnable in my catbird library). You really should just stick with cats.effect.IO, though, and use the provided conversions to convert to and from futures in any parts of your code where you're working with traditional Future-based APIs.
