RxScala: Add toFuture to BlockingObservable

Author: zsxwing

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/observables/BlockingObservable.scala

@@ -16,6 +16,7 @@
 package rx.lang.scala.observables
 
 import scala.collection.JavaConverters._
+import scala.concurrent.{Future, Promise}
 import rx.lang.scala.ImplicitFunctionConversions._
 import rx.lang.scala.Observable
 import rx.observables.{BlockingObservable => JBlockingObservable}
@@ -227,8 +228,6 @@ class BlockingObservable[+T] private[scala] (val o: Observable[T])
     }
   }
 
-  // TODO toFuture()
-
   /**
    * Returns an {@link Iterator} that iterates over all items emitted by this {@link Observable}.
    */
@@ -258,6 +257,23 @@ class BlockingObservable[+T] private[scala] (val o: Observable[T])
     asJava.latest.asScala: Iterable[T] // useless ascription because of compiler bug
   }
 
+  /**
+   * Returns a `Future` representing the single value emitted by this `BlockingObservable`.
+   *
+   * `toFuture` throws an `IllegalArgumentException` if the `BlockingObservable` emits more than one item. If the
+   * `BlockingObservable` may emit more than item, use `BlockingObservable.toList.toFuture`.
+   *
+   * `toFuture` throws an `NoSuchElementException` if the `BlockingObservable` is empty.
+   *
+   * <img width="640" height="395" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.toFuture.png">
+   *
+   * @return a `Future` that expects a single item to be emitted by this `BlockingObservable`.
+   */
+  def toFuture: Future[T] = {
+    val p = Promise[T]()
+    o.single.subscribe(t => p.success(t), e => p.failure(e))
+    p.future
+  }
 }
 
 // Cannot yet have inner class because of this error message: 

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/observables/BlockingObservableTest.scala

@@ -1,5 +1,22 @@
+/**
+ * Copyright 2013 Netflix, 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 rx.lang.scala.observables
 
+import scala.concurrent.Await
+import scala.concurrent.duration._
 import org.junit.Assert._
 import org.junit.Test
 import org.scalatest.junit.JUnitSuite
@@ -113,4 +130,23 @@ class BlockingObservableTest extends JUnitSuite {
     val o = Observable.items(1, 2)
     assertEquals(2, o.toBlocking.lastOrElse(3))
   }
+
+  @Test
+  def testToFuture() {
+    val o = Observable.items(1)
+    val r = Await.result(o.toBlocking.toFuture, 10 seconds)
+    assertEquals(1, r)
+  }
+
+  @Test(expected = classOf[NoSuchElementException])
+  def testToFutureWithEmpty() {
+    val o = Observable.empty
+    Await.result(o.toBlocking.toFuture, 10 seconds)
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testToFutureWithMultipleItems() {
+    val o = Observable.items(1, 2)
+    Await.result(o.toBlocking.toFuture, 10 seconds)
+  }
 }