Add unsafeRunToTwitterFuture for twitter Future

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

@@ -16,8 +16,9 @@
 
 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 {
@@ -33,4 +34,23 @@ package object twitter {
         }
       }
   }
+
+  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,22 +2,32 @@ package scalaz.zio.interop
 
 import java.util.concurrent.atomic.AtomicInteger
 
-import com.twitter.util.{ Future, JavaTimer, Duration => TwitterDuration }
+import com.twitter.util.{ Await, Future, JavaTimer, TimeoutException, Timer, Duration => TwitterDuration }
 import org.specs2.concurrent.ExecutionEnv
+import org.specs2.specification.AfterAll
 import scalaz.zio.Exit.Cause.Fail
 import scalaz.zio.duration._
 import scalaz.zio.interop.twitter._
 import scalaz.zio.{ FiberFailure, Task, TestRuntime, ZIO }
 
-class TwitterSpec(implicit ee: ExecutionEnv) extends TestRuntime {
+class TwitterSpec(implicit ee: ExecutionEnv) extends TestRuntime with AfterAll {
   def is =
     "Twitter spec".title ^ s2"""
     `Task.fromTwitterFuture` must
       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 implicit val twitterTimer: Timer = new JavaTimer(true)
+
+  override def afterAll: Unit = twitterTimer.stop()
+
   private def propagateFailures = {
     val error  = new Exception
     val future = Task(Future.exception[Int](error))
@@ -35,8 +45,6 @@ class TwitterSpec(implicit ee: ExecutionEnv) extends TestRuntime {
   }
 
   private def propagateInterrupts = {
-    implicit val timer = new JavaTimer(true)
-
     val value       = new AtomicInteger(0)
     val futureDelay = TwitterDuration.fromSeconds(1)
     val future      = Task.succeed(Future.sleep(futureDelay).map(_ => value.incrementAndGet()))
@@ -47,4 +55,26 @@ class TwitterSpec(implicit ee: ExecutionEnv) extends TestRuntime {
     unsafeRun(task <* ZIO.unit.delay(3.seconds)) must beNone
     value.get() ==== 0
   }
+
+  private def evaluateToSuccessfulFuture = {
+    Await.result(this.unsafeRunToTwitterFuture(Task.succeed(1))) ==== 1
+    Await.result(this.unsafeRunToTwitterFuture(Task.succeed(2).delay(100.millis))) ==== 2
+  }
+
+  private def evaluateToFailedFuture = {
+    val e    = new Exception
+    val task = ZIO.unit.delay(100.millis) *> Task.fail(e).unit
+
+    Await.result(this.unsafeRunToTwitterFuture(task)) must throwAn(e)
+  }
+
+  private def evaluateToInterruptedFuture = {
+    val value         = new AtomicInteger(0)
+    val futureTimeout = TwitterDuration.fromMilliseconds(100)
+
+    val task = ZIO.unit.delay(500.millis) *> Task.effect(value.incrementAndGet())
+
+    Await.result(this.unsafeRunToTwitterFuture(task).raiseWithin(futureTimeout)) must throwAn[TimeoutException]
+    unsafeRun(ZIO.unit.delay(600.millis) *> Task.effect(value.get)) ==== 0
+  }
 }