Add unsafeRunToTwitterFuture for twitter Future

interop-twitter/jvm/src/main/scala/scalaz/zio/interop/twitter.scala

@@ -16,21 +16,41 @@
 
 package scalaz.zio.interop
 
-import com.twitter.util.{ Future, FutureCancelledException, Return, Throw }
-import scalaz.zio.{ Task, UIO }
+import com.twitter.util.{ Future, FutureCancelledException, Promise, Return, Throw }
+import scalaz.zio.Exit.Cause
+import scalaz.zio.{ Runtime, Task, UIO, ZIO }
 
 package object twitter {
   implicit class TaskObjOps(private val obj: Task.type) extends AnyVal {
     final def fromTwitterFuture[A](future: Task[Future[A]]): Task[A] =
-      future.flatMap { f =>
-        Task.effectAsyncInterrupt { cb =>
-          f.respond {
-            case Return(a) => cb(Task.succeed(a))
-            case Throw(e)  => cb(Task.fail(e))
-          }
-
-          Left(UIO(f.raise(new FutureCancelledException)))
+      Task.uninterruptibleMask { restore =>
+        future.flatMap { f =>
+          restore(Task.effectAsync { cb: (Task[A] => Unit) =>
+            val _ = f.respond {
+              case Return(a) => cb(Task.succeed(a))
+              case Throw(e)  => cb(Task.fail(e))
+            }
+          }).onInterrupt(UIO(f.raise(new FutureCancelledException)))
         }
       }
   }
+
+  implicit class RuntimeOps[R](private val runtime: Runtime[R]) extends AnyVal {
+    def unsafeRunToTwitterFuture[A](io: ZIO[R, Throwable, A]): Future[A] = {
+      val promise                                = Promise[A]()
+      def failure(cause: Cause[Throwable]): Unit = promise.setException(cause.squash)
+      def success(value: A): Unit                = promise.setValue(value)
+
+      runtime.unsafeRunAsync {
+        io.fork.flatMap { f =>
+          promise.setInterruptHandler {
+            case _ => runtime.unsafeRunAsync_(f.interrupt)
+          }
+          f.join
+        }
+      }(_.fold(failure, success))
+
+      promise
+    }
+  }
 }

interop-twitter/jvm/src/test/scala/scalaz/zio/interop/TwitterSpec.scala

@@ -2,10 +2,9 @@ package scalaz.zio.interop
 
 import java.util.concurrent.atomic.AtomicInteger
 
-import com.twitter.util.{ Future, JavaTimer, Duration => TwitterDuration }
+import com.twitter.util.{ Await, Future, Promise }
 import org.specs2.concurrent.ExecutionEnv
 import scalaz.zio.Exit.Cause.Fail
-import scalaz.zio.duration._
 import scalaz.zio.interop.twitter._
 import scalaz.zio.{ FiberFailure, Task, TestRuntime, ZIO }
 
@@ -16,6 +15,11 @@ class TwitterSpec(implicit ee: ExecutionEnv) extends TestRuntime {
       return failing `Task` if future failed.          $propagateFailures
       return successful `Task` if future succeeded.    $propagateResults
       ensure future is interrupted together with task. $propagateInterrupts
+
+    `Runtime.unsafeRunToTwitterFuture` must
+      return successful `Future` if Task evaluation succeeded.    $evaluateToSuccessfulFuture
+      return failed `Future` if Task evaluation failed.           $evaluateToFailedFuture
+      ensure Task evaluation is interrupted together with Future. $evaluateToInterruptedFuture
     """
 
   private def propagateFailures = {
@@ -35,16 +39,45 @@ class TwitterSpec(implicit ee: ExecutionEnv) extends TestRuntime {
   }
 
   private def propagateInterrupts = {
-    implicit val timer = new JavaTimer(true)
+    val value = new AtomicInteger(0)
+
+    val promise = new Promise[Unit] with Promise.InterruptHandler {
+      override protected def onInterrupt(t: Throwable): Unit = setException(t)
+    }
+
+    val future = Task(promise.flatMap(_ => Future(value.incrementAndGet())))
+
+    unsafeRun {
+      for {
+        fiber <- Task.fromTwitterFuture(future).fork
+        _     <- fiber.interrupt
+        _     <- Task.effect(promise.setDone())
+        a     <- fiber.await
+      } yield (a.toEither must beLeft) and (value.get ==== 0)
+    }
+  }
+
+  private def evaluateToSuccessfulFuture =
+    Await.result(this.unsafeRunToTwitterFuture(Task.succeed(2))) ==== 2
+
+  private def evaluateToFailedFuture = {
+    val e    = new Exception
+    val task = Task.fail(e).unit
+
+    Await.result(this.unsafeRunToTwitterFuture(task)) must throwAn(e)
+  }
 
-    val value       = new AtomicInteger(0)
-    val futureDelay = TwitterDuration.fromSeconds(1)
-    val future      = Task.succeed(Future.sleep(futureDelay).map(_ => value.incrementAndGet()))
+  private def evaluateToInterruptedFuture = {
+    val value = new AtomicInteger(0)
 
-    val taskTimeout = 500.millis
-    val task        = Task.fromTwitterFuture(future).timeout(taskTimeout)
+    val task: ZIO[Any, Throwable, Future[Int]] = for {
+      promise <- scalaz.zio.Promise.make[Throwable, Int]
+      t       = promise.await.flatMap(_ => Task.effectTotal(value.incrementAndGet()))
+      future  = this.unsafeRunToTwitterFuture(t)
+      _       = future.raise(new Exception)
+      _       <- promise.succeed(1)
+    } yield future
 
-    unsafeRun(task <* ZIO.unit.delay(3.seconds)) must beNone
-    value.get() ==== 0
+    (Await.result(unsafeRun(task)) must throwAn[InterruptedException]) and (value.get ==== 0)
   }
 }