add Observable.from(Future) constructor

Author: samuelgruetter

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -1828,8 +1828,11 @@ object Observable {
   import scala.collection.JavaConverters._
   import scala.collection.immutable.Range
   import scala.concurrent.duration.Duration
+  import scala.concurrent.{Future, ExecutionContext}
+  import scala.util.{Success, Failure}
   import ImplicitFunctionConversions._
   import JavaConversions._
+  import rx.lang.scala.subjects.AsyncSubject
 
   private[scala]
   def jObsOfListToScObsOfSeq[T](jObs: rx.Observable[_ <: java.util.List[T]]): Observable[Seq[T]] = {
@@ -1911,6 +1914,24 @@ object Observable {
     toScalaObservable[T](rx.Observable.from(items.toIterable.asJava))
   }
 
+ /** Returns an Observable emitting the value produced by the Future as its single item.
+   * If the future fails, the Observable will fail as well.
+   *
+   * @param f Future whose value ends up in the resulting Observable
+   * @return an Observable completed after producing the value of the future, or with an exception
+   */
+  def from[T](f: Future[T])(implicit execContext: ExecutionContext): Observable[T] = {
+    val s = AsyncSubject[T]()
+    f.onComplete {
+      case Failure(e) =>
+        s.onError(e)
+      case Success(c) =>
+        s.onNext(c)
+        s.onCompleted()
+    }
+    s
+  }
+
   /**
    * Converts an `Iterable` into an Observable.
    *

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/ObservableTest.scala

@@ -1,5 +1,8 @@
 package rx.lang.scala
 
+import scala.concurrent.{Future, Await}
+import scala.concurrent.duration.Duration
+import scala.concurrent.ExecutionContext.Implicits.global
 import org.junit.Assert._
 import org.junit.{ Ignore, Test }
 import org.scalatest.junit.JUnitSuite
@@ -67,6 +70,30 @@ class ObservableTests extends JUnitSuite {
     assertEquals(receivedMsg, msg)
   }
 
+  @Test def testFromFuture() {
+    val o = Observable from Future { 5 }
+    assertEquals(5, o.toBlockingObservable.single)
+  }
+
+  @Test def testFromFutureWithDelay() {
+    val o = Observable from Future { Thread.sleep(200); 42 }
+    assertEquals(42, o.toBlockingObservable.single)
+  }
+
+  @Test def testFromFutureWithError() {
+    val err = new Exception("ooops42")
+    val o: Observable[Int] = Observable from Future { Thread.sleep(200); throw err }
+    assertEquals(List(Notification.OnError(err)), o.materialize.toBlockingObservable.toList)
+  }
+
+  @Test def testFromFutureWithSubscribeOnlyAfterCompletion() {
+    val f = Future { Thread.sleep(200); 6 }
+    val o = Observable from f
+    val res = Await.result(f, Duration.Inf)
+    assertEquals(6, res)
+    assertEquals(6, o.toBlockingObservable.single)
+  }
+
   /*
  @Test def testHead() {
    val observer = mock(classOf[Observer[Int]])