JavaFuture <=> TwitterFuture bijection

bijection-util/src/main/scala/com/twitter/bijection/twitter_util/JavaFutureToTwitterFutureConverter.scala

@@ -0,0 +1,107 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package com.twitter.bijection.twitter_util
+
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ Future => JFuture }
+
+import com.twitter.util.{ Future, Promise, Try }
+
+import scala.annotation.tailrec
+
+/**
+ * Utility class for converting Java futures to Twitter's
+ *
+ * @param waitTimeMs Time spent sleeping by the thread converting java futures to
+ *                   twitter futures when there are no futures to convert in ms
+ */
+private[twitter_util] class JavaFutureToTwitterFutureConverter(waitTimeMs: Long = 1000L) {
+
+  case class Link[T](future: JFuture[T], promise: Promise[T]) {
+    def maybeUpdate: Boolean = future.isDone && {
+      promise.update(Try(future.get()))
+      true
+    }
+
+    def cancel(): Unit = {
+      promise.setException(new Exception("Promise not completed"))
+      future.cancel(true)
+    }
+  }
+
+  sealed abstract class State
+  case object Closed extends State
+  case class Open(links: List[Link[_]]) extends State
+  val EmptyState: State = Open(Nil)
+
+  private val pollRun = new Runnable {
+    override def run(): Unit =
+      try {
+        if (!Thread.currentThread().isInterrupted)
+          loop(list.getAndSet(EmptyState))
+      } catch {
+        case e: InterruptedException =>
+      }
+
+    @tailrec
+    def swapOpen(old: List[Link[_]]): Option[List[Link[_]]] = list.get match {
+      case Closed => None
+      case s @ Open(links) =>
+        if (list.compareAndSet(s, Open(old))) Some(links)
+        else swapOpen(links)
+    }
+
+    @tailrec
+    def loop(state: State): Unit = state match {
+      case s @ Open(links) =>
+        val notDone = links.filterNot(_.maybeUpdate)
+        if (links.isEmpty || notDone.nonEmpty) Thread.sleep(waitTimeMs)
+        swapOpen(notDone) match {
+          case None => notDone.foreach(_.cancel())
+          case Some(next) => loop(Open(next))
+        }
+      case Closed =>
+    }
+  }
+  private val list = new AtomicReference[State](EmptyState)
+  private val thread = new Thread(pollRun)
+
+  def apply[T](javaFuture: JFuture[T]): Future[T] = {
+    val promise = new Promise[T]()
+    poll(Link(javaFuture, promise))
+    promise
+  }
+
+  def start(): Unit = {
+    thread.setDaemon(true)
+    thread.start()
+  }
+
+  def stop(): Unit = {
+    list.getAndSet(Closed) match {
+      case Closed => // already closed
+      case s @ Open(links) => links.foreach(_.cancel())
+    }
+  }
+
+  private def poll[T](link: Link[T]): Unit = list.get match {
+    case Closed => link.cancel()
+    case s @ Open(tail) =>
+      if (list.compareAndSet(s, Open(link :: tail))) ()
+      else poll(link)
+  }
+}

bijection-util/src/main/scala/com/twitter/bijection/twitter_util/UtilBijections.scala

@@ -16,7 +16,9 @@ limitations under the License.
 
 package com.twitter.bijection.twitter_util
 
-import com.twitter.bijection.{ AbstractBijection, Bijection, ImplicitBijection }
+import java.util.concurrent.{ Future => JavaFuture }
+
+import com.twitter.bijection._
 import com.twitter.io.Buf
 import com.twitter.util.{ Future => TwitterFuture, Try => TwitterTry, Promise => TwitterPromise, Return, Throw, FuturePool }
 
@@ -40,7 +42,7 @@ trait UtilBijections {
   implicit def futureBijection[A, B](implicit bij: ImplicitBijection[A, B]): Bijection[TwitterFuture[A], TwitterFuture[B]] =
     new AbstractBijection[TwitterFuture[A], TwitterFuture[B]] {
       override def apply(fa: TwitterFuture[A]) = fa.map(bij(_))
-      override def invert(fb: TwitterFuture[B]) = fb.map(bij.invert(_))
+      override def invert(fb: TwitterFuture[B]) = fb.map(bij.invert)
     }
 
   /**
@@ -50,7 +52,7 @@ trait UtilBijections {
   implicit def futureScalaBijection[A, B](implicit bij: ImplicitBijection[A, B], executor: ExecutionContext): Bijection[ScalaFuture[A], ScalaFuture[B]] =
     new AbstractBijection[ScalaFuture[A], ScalaFuture[B]] {
       override def apply(fa: ScalaFuture[A]) = fa.map(bij(_))
-      override def invert(fb: ScalaFuture[B]) = fb.map(bij.invert(_))
+      override def invert(fb: ScalaFuture[B]) = fb.map(bij.invert)
     }
 
   /**
@@ -78,6 +80,22 @@ trait UtilBijections {
     }
   }
 
+  /**
+   * Injection from java futures to twitter futures
+   */
+  implicit def twitter2JavaFuture[A](
+    implicit converter: JavaFutureToTwitterFutureConverter
+  ): Bijection[TwitterFuture[A], JavaFuture[A]] = {
+    new AbstractBijection[TwitterFuture[A], JavaFuture[A]] {
+      override def apply(f: TwitterFuture[A]): JavaFuture[A] =
+        f.toJavaFuture.asInstanceOf[JavaFuture[A]]
+
+      override def invert(f: JavaFuture[A]): TwitterFuture[A] = {
+        converter(f)
+      }
+    }
+  }
+
   /**
    * Bijection between twitter and scala style Trys
    */
@@ -102,7 +120,7 @@ trait UtilBijections {
   implicit def tryBijection[A, B](implicit bij: ImplicitBijection[A, B]): Bijection[TwitterTry[A], TwitterTry[B]] =
     new AbstractBijection[TwitterTry[A], TwitterTry[B]] {
       override def apply(fa: TwitterTry[A]) = fa.map(bij(_))
-      override def invert(fb: TwitterTry[B]) = fb.map(bij.invert(_))
+      override def invert(fb: TwitterTry[B]) = fb.map(bij.invert)
     }
 
   /**
@@ -112,7 +130,7 @@ trait UtilBijections {
   implicit def tryScalaBijection[A, B](implicit bij: ImplicitBijection[A, B]): Bijection[ScalaTry[A], ScalaTry[B]] =
     new AbstractBijection[ScalaTry[A], ScalaTry[B]] {
       override def apply(fa: ScalaTry[A]) = fa.map(bij(_))
-      override def invert(fb: ScalaTry[B]) = fb.map(bij.invert(_))
+      override def invert(fb: ScalaTry[B]) = fb.map(bij.invert)
     }
 
   /**

bijection-util/src/test/scala/com/twitter/bijection/twitter_util/JavaFutureToTwitterFutureConverterSpec.scala

@@ -0,0 +1,53 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package com.twitter.bijection.twitter_util
+
+import java.util.concurrent.{ Callable, FutureTask }
+
+import com.twitter.util.Await
+import org.scalatest.{ Matchers, WordSpec }
+
+class JavaFutureToTwitterFutureConverterSpec extends WordSpec with Matchers {
+  trait Fixtures {
+    val converter = {
+      val c = new JavaFutureToTwitterFutureConverter
+      c.start()
+      c
+    }
+  }
+
+  "JavaFutureToTwitterFutureConverter" should {
+    "do a proper round trip between a java future to a twitter future" in new Fixtures {
+      val jFuture = new FutureTask[Int](new Callable[Int] {
+        override def call(): Int = 3
+      })
+      val tFuture = converter(jFuture)
+      tFuture.toJavaFuture === jFuture
+    }
+    "cancel the future when calling apply after stop" in new Fixtures {
+      val jFuture = new FutureTask[Int](new Callable[Int] {
+        override def call(): Int = 3
+      })
+      converter.stop()
+      val tFuture = converter(jFuture)
+      the[Exception] thrownBy {
+        Await.result(tFuture)
+      } should have message "Promise not completed"
+      jFuture.isCancelled shouldBe true
+    }
+  }
+}

bijection-util/src/test/scala/com/twitter/bijection/twitter_util/UtilBijectionLaws.scala

@@ -16,72 +16,92 @@ limitations under the License.
 
 package com.twitter.bijection.twitter_util
 
-import com.twitter.bijection.{ CheckProperties, BaseProperties, Bijection }
+import com.google.common.util.concurrent.Futures
+import com.twitter.bijection.{ CheckProperties, BaseProperties }
 import com.twitter.io.Buf
 import com.twitter.util.{ Future => TwitterFuture, Try => TwitterTry, Await => TwitterAwait }
 import java.lang.{ Integer => JInt, Long => JLong }
+import java.util.concurrent.{ Future => JavaFuture }
 import org.scalacheck.Arbitrary
-import org.scalatest.{ PropSpec, MustMatchers }
-import org.scalatest.prop.PropertyChecks
+import org.scalatest.BeforeAndAfterAll
 
-import org.scalacheck.Prop.forAll
 import scala.concurrent.{ Future => ScalaFuture, Await => ScalaAwait }
 import scala.concurrent.duration.Duration
 import scala.util.{ Try => ScalaTry }
 import scala.concurrent.future
 import scala.concurrent.ExecutionContext.Implicits.global
 
-class UtilBijectionLaws extends CheckProperties with BaseProperties {
+class UtilBijectionLaws extends CheckProperties with BaseProperties with BeforeAndAfterAll {
   import UtilBijections._
 
-  protected def toOption[T](f: TwitterFuture[T]): Option[T] = TwitterTry(TwitterAwait.result(f)).toOption
+  protected def toOption[T](f: TwitterFuture[T]): Option[T] =
+    TwitterTry(TwitterAwait.result(f)).toOption
 
-  protected def toOption[T](f: ScalaFuture[T]): Option[T] = TwitterTry(ScalaAwait.result(f, Duration.Inf)).toOption
+  protected def toOption[T](f: ScalaFuture[T]): Option[T] =
+    TwitterTry(ScalaAwait.result(f, Duration.Inf)).toOption
 
-  implicit def futureArb[T: Arbitrary] = arbitraryViaFn[T, TwitterFuture[T]] { TwitterFuture.value(_) }
+  protected def toOption[T](f: JavaFuture[T]): Option[T] =
+    TwitterTry(f.get()).toOption
+
+  implicit def futureArb[T: Arbitrary] = arbitraryViaFn[T, TwitterFuture[T]] { TwitterFuture.value }
   implicit def scalaFutureArb[T: Arbitrary] = arbitraryViaFn[T, ScalaFuture[T]] { future(_) }
+  implicit def javaFutureArb[T: Arbitrary] =
+    arbitraryViaFn[T, JavaFuture[T]] { Futures.immediateFuture }
   implicit def tryArb[T: Arbitrary] = arbitraryViaFn[T, TwitterTry[T]] { TwitterTry(_) }
   implicit def scalaTryArb[T: Arbitrary] = arbitraryViaFn[T, ScalaTry[T]] { ScalaTry(_) }
 
   implicit val jIntArb = arbitraryViaBijection[Int, JInt]
   implicit val jLongArb = arbitraryViaBijection[Long, JLong]
   implicit val bufArb: Arbitrary[Buf] = arbitraryViaFn[Array[Byte], Buf](Buf.ByteArray.Owned.apply)
 
-  implicit protected def futureEq[T: Equiv]: Equiv[TwitterFuture[T]] = Equiv.fromFunction { (f1, f2) =>
-    Equiv[Option[T]].equiv(toOption(f1), toOption(f2))
+  implicit val futureConverter = {
+    val converter = new JavaFutureToTwitterFutureConverter()
+    converter.start()
+    converter
   }
 
-  implicit protected def scalaFutureEq[T: Equiv]: Equiv[ScalaFuture[T]] = Equiv.fromFunction { (f1, f2) =>
-    Equiv[Option[T]].equiv(toOption(f1), toOption(f2))
-  }
+  override protected def afterAll(): Unit = futureConverter.stop()
+
+  implicit protected def futureEq[T: Equiv]: Equiv[TwitterFuture[T]] =
+    Equiv.fromFunction { (f1, f2) => Equiv[Option[T]].equiv(toOption(f1), toOption(f2)) }
+
+  implicit protected def scalaFutureEq[T: Equiv]: Equiv[ScalaFuture[T]] =
+    Equiv.fromFunction { (f1, f2) => Equiv[Option[T]].equiv(toOption(f1), toOption(f2)) }
+
+  implicit protected def javaFutureEq[T: Equiv]: Equiv[JavaFuture[T]] =
+    Equiv.fromFunction { (f1, f2) => Equiv[Option[T]].equiv(toOption(f1), toOption(f2)) }
 
   type FromMap = Map[Int, Long]
   type ToMap = Map[JInt, JLong]
 
-  property("round trips com.twitter.util.Future[Map[Int, String]] -> com.twitter.util.Future[JInt, JLong]") {
+  property("round trips TwitterFuture[Map[Int, String]] -> Twitter.Future[JInt, JLong]") {
     isBijection[TwitterFuture[FromMap], TwitterFuture[ToMap]]
   }
 
-  property("round trips scala.concurrent.Future[Map[Int, String]] -> scala.concurrent.Future[JInt, JLong]") {
+  property("round trips ScalaFuture[Map[Int, String]] -> ScalaFuture[JInt, JLong]") {
     isBijection[ScalaFuture[FromMap], ScalaFuture[ToMap]]
   }
 
-  property("round trips com.twitter.util.Try[Map[Int, String]] -> com.twitter.util.Try[Map[JInt, JLong]]") {
+  property("round trips TwitterTry[Map[Int, String]] -> TwitterTry[Map[JInt, JLong]]") {
     isBijection[TwitterTry[FromMap], TwitterTry[ToMap]]
   }
 
-  property("round trips scala.util.Try[Map[Int, String]] -> scala.util.Try[Map[JInt, JLong]]") {
+  property("round trips ScalaTry[Map[Int, String]] -> ScalaTry[Map[JInt, JLong]]") {
     isBijection[ScalaTry[FromMap], ScalaTry[ToMap]]
   }
 
-  property("round trips com.twitter.util.Try[Map[JInt, JLong]] -> scala.util.Try[Map[JInt, JLong]]") {
+  property("round trips TwitterTry[Map[JInt, JLong]] -> ScalaTry[Map[JInt, JLong]]") {
     isBijection[TwitterTry[ToMap], ScalaTry[ToMap]]
   }
 
-  property("round trips com.twitter.util.Future[Map[JInt, JLong]] -> scala.concurrent.Future[Map[JInt, JLong]]") {
+  property("round trips TwitterFuture[Map[JInt, JLong]] -> ScalaFuture[Map[JInt, JLong]]") {
     isBijection[TwitterFuture[ToMap], ScalaFuture[ToMap]]
   }
 
+  property("round trips TwitterFuture[Map[JInt, JLong]] -> JavaFuture[Map[JInt, JLong]]") {
+    isBijection[TwitterFuture[ToMap], JavaFuture[ToMap]]
+  }
+
   property("round trips shared com.twitter.io.Buf -> Array[Byte]") {
     import Shared.byteArrayBufBijection
 

build.sbt

@@ -242,7 +242,10 @@ lazy val bijectionJson = module("json").settings(
 
 lazy val bijectionUtil = module("util").settings(
   osgiExportAll("com.twitter.bijection.twitter_util"),
-  libraryDependencies += "com.twitter" %% "util-core" % "6.24.0"
+  libraryDependencies ++= Seq(
+    "com.twitter" %% "util-core" % "6.24.0",
+    "com.google.guava" % "guava" % "14.0" % "test"
+  )
 ).dependsOn(bijectionCore % "test->test;compile->compile")
 
 lazy val bijectionFinagleMySql = module("finagle-mysql").settings(