Add IO#to conversion method

core/shared/src/main/scala/cats/effect/Effect.scala

@@ -52,4 +52,10 @@ trait Effect[F[_]] extends Async[F] with LiftIO[F] {
       })
     }
   }
+
+  override def liftIO[A](ioa: IO[A]): F[A] = {
+    // Implementation for `IO#to` depends on the `Async` type class,
+    // and not on `Effect`, so this shouldn't create a cyclic dependency
+    ioa.to[F](this)
+  }
 }

core/shared/src/main/scala/cats/effect/IO.scala

@@ -19,6 +19,7 @@ package effect
 
 import cats.effect.internals.{AndThen, IOPlatform, NonFatal}
 import scala.annotation.tailrec
+import scala.annotation.unchecked.uncheckedVariance
 import scala.concurrent.{ExecutionContext, Future, Promise}
 import scala.concurrent.duration._
 import scala.util.{Left, Right}
@@ -380,6 +381,23 @@ sealed abstract class IO[+A] {
     p.future
   }
 
+  /**
+   * Converts the source `IO` into any `F` type that implements
+   * the [[cats.effect.Async Async]] type class.
+   */
+  final def to[F[_]](implicit F: cats.effect.Async[F]): F[A @uncheckedVariance] =
+    this match {
+      case Pure(a) => F.pure(a)
+      case RaiseError(e) => F.raiseError(e)
+      case Suspend(thunk) => F.suspend(thunk(()).to[F])
+      case Async(k) => F.async(k)
+
+      case BindSuspend(thunk, f) =>
+        F.flatMap(F.suspend(thunk(()).to[F]))(e => f(e).to[F])
+      case BindAsync(k, f) =>
+        F.flatMap(F.async(k))(e => f(e).to[F])
+    }
+
   override def toString = this match {
     case Pure(a) => s"IO($a)"
     case RaiseError(e) => s"IO(throw $e)"
@@ -426,7 +444,7 @@ private[effect] trait IOInstances extends IOLowPriorityInstances {
 
     override def shift[A](ioa: IO[A])(implicit ec: ExecutionContext) = ioa.shift
 
-    def liftIO[A](ioa: IO[A]) = ioa
+    override def liftIO[A](ioa: IO[A]) = ioa
   }
 
   implicit def ioMonoid[A: Monoid]: Monoid[IO[A]] = new IOSemigroup[A] with Monoid[IO[A]] {
@@ -591,7 +609,7 @@ object IO extends IOInstances {
 
   private final case class Pure[+A](a: A)
     extends IO[A]
-  private final case class RaiseError(t: Throwable)
+  private final case class RaiseError(e: Throwable)
     extends IO[Nothing]
   private final case class Suspend[+A](thunk: AndThen[Unit, IO[A]])
     extends IO[A]

core/shared/src/test/scala/cats/effect/Generators.scala

@@ -36,6 +36,14 @@ object Generators {
       10 -> genFlatMap[A])
   }
 
+  def genSyncIO[A: Arbitrary: Cogen]: Gen[IO[A]] = {
+    Gen.frequency(
+      5 -> genPure[A],
+      5 -> genApply[A],
+      1 -> genFail[A],
+      10 -> genFlatMap[A])
+  }
+
   def genPure[A: Arbitrary]: Gen[IO[A]] = arbitrary[A].map(IO.pure(_))
 
   def genApply[A: Arbitrary]: Gen[IO[A]] = arbitrary[A].map(IO.apply(_))

laws/shared/src/test/scala/cats/effect/IOTests.scala

@@ -351,12 +351,48 @@ class IOTests extends BaseTestsSuite {
     val f = io.unsafeToFuture(); ec.tick()
     f.value shouldEqual Some(Success(100))
   }
+
+  def repeatedTransformLoop[A](n: Int, io: IO[A]): IO[A] =
+    io.to[IO].flatMap { x =>
+      if (n <= 0) IO.pure(x).to[IO]
+      else repeatedTransformLoop(n - 1, io)
+    }
+
+  testAsync("io.to[IO] <-> io") { implicit ec =>
+    check { (io: IO[Int]) => io.to[IO] <-> io }
+  }
+
+  testAsync("sync.to[IO] is stack-safe") { implicit ec =>
+    // Override default generator to only generate
+    // synchronous instances that are stack-safe
+    implicit val arbIO: Arbitrary[IO[Int]] =
+      Arbitrary(Gen.delay(genSyncIO[Int]))
+
+    check { (io: IO[Int]) =>
+      repeatedTransformLoop(10000, io) <-> io
+    }
+  }
+
+  testAsync("async.to[IO] is stack-safe if the source is") { implicit ec =>
+    // Stack-safe async IO required
+    def async(a: Int) = IO.async[Int] { cb =>
+      ec.execute(new Runnable {
+        def run(): Unit =
+          cb(Right(a))
+      })
+    }
+
+    val f = repeatedTransformLoop(10000, async(99)).unsafeToFuture()
+    ec.tick()
+    f.value shouldEqual Some(Success(99))
+  }
 }
 
 object IOTests {
   /** Implementation for testing default methods. */
   val ioEffectDefaults = new Effect[IO] {
     private val ref = implicitly[Effect[IO]]
+
     def async[A](k: ((Either[Throwable, A]) => Unit) => Unit): IO[A] =
       ref.async(k)
     def raiseError[A](e: Throwable): IO[A] =
@@ -373,7 +409,5 @@ object IOTests {
       ref.runAsync(fa)(cb)
     def suspend[A](thunk: =>IO[A]): IO[A] =
       ref.suspend(thunk)
-    def liftIO[A](ioa: IO[A]): IO[A] =
-      ref.liftIO(ioa)
   }
 }