Effect[Future]

README.md

Instances of cats Effect for Future

Future is not a Monad we all know that, but sometimes we have Java developers programming in Scala. This implementation can help the effort to migrate legacy code to Tagless Final.

implicit def futureEffect: Effect[Future] = new FutureEffect()

implicit def futureEffect: ConcurrentEffect[Future] = new FutureConcurrentEffect()

FutureConcurrentEffect.scala

import cats.effect._

import scala.concurrent.{ExecutionContext, Future}

class FutureConcurrentEffect()(implicit ec: ExecutionContext) extends FutureEffect with ConcurrentEffect[Future] {
  def start[A](fa: Future[A]): Future[Fiber[Future, A]] = Future.successful {
    FutureFiber(fa)
  }

  def racePair[A, B](fa: Future[A], fb: Future[B]): Future[Either[(A, Fiber[Future, B]), (Fiber[Future, A], B)]] = ???

  def runCancelable[A](fa: Future[A])(cb: Either[Throwable, A] => IO[Unit]): SyncIO[CancelToken[Future]] = ???
}

FutureEffect.scala

import cats.effect._

import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Failure, Success}

class FutureEffect()(implicit ec: ExecutionContext) extends Effect[Future] {
  def runAsync[A](fa: Future[A])(cb: Either[Throwable, A] => IO[Unit]): SyncIO[Unit] = SyncIO {
    fa.onComplete(_.fold(fa => cb(Left(fa)), fb => cb(Right(fb))))
  }

  def async[A](k: (Either[Throwable, A] => Unit) => Unit): Future[A] = {
    val p = Promise[A]()
    k(_.fold(p.failure, p.success))
    p.future
  }

  def asyncF[A](k: (Either[Throwable, A] => Unit) => Future[Unit]): Future[A] = {
    val p = Promise[A]()
    k(_.fold(p.failure, p.success))
    p.future
  }

  def suspend[A](thunk: => Future[A]): Future[A] =
    Future.successful(()).flatMap(_ => thunk)

  def bracketCase[A, B](acquire: Future[A])
                       (use: A => Future[B])
                       (release: (A, ExitCase[Throwable]) => Future[Unit]): Future[B] = for {
    a <- acquire
    etb <- use(a).transformWith[Either[Throwable, B]] {
      case Success(v) => pure(Right(v))
      case Failure(e) => pure(Left(e))
    }
    _ <- release(a, etb match {
      case Right(_) => ExitCase.complete
      case Left(t) => ExitCase.error(t)
    })
    e <- etb.fold(raiseError, pure)
  } yield e

  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] = flatMap(f(a)) {
    case Left(a) => tailRecM(a)(f)
    case Right(b) => pure(b)
  }

  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 e: Throwable => f(e)
  }
}

FutureFiber.scala

import cats.effect.{CancelToken, Fiber}

import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContext, Future, Promise}

final case class FutureFiber[A](fa: Future[A])(implicit ec: ExecutionContext) extends Fiber[Future, A] {
  val p = Promise[A]()
  fa.onComplete(_.fold(p.failure, p.success))

  def cancel: CancelToken[Future] = Future.successful {
    p.failure(new InterruptedException())
    ()
  }

  def join: Future[A] = Await.ready(p.future, Duration.Inf)
}