Add FutureToObservable and rename ObservableExtensions to IterableToObservable

Author: David Hoepelman

src/main/scala/rx/lang/scala/package.scala

@@ -15,7 +15,8 @@
  */
 package rx.lang
 
-import _root_.scala.util.{Try, Success, Failure}
+import _root_.scala.util.Try
+import _root_.scala.concurrent.{ExecutionContext, Future}
 
 /**
  * This package contains all classes that RxScala users need.
@@ -24,11 +25,11 @@ import _root_.scala.util.{Try, Success, Failure}
  */
 package object scala {
 
-  /**
-   * Placeholder for extension methods into Observable[T] from other types
-   */
-  implicit class ObservableExtensions[T](val source: Iterable[T]) extends AnyVal {
-      def toObservable: Observable[T] = {  Observable.from(source) }
+  @deprecated("Use IterableToObservable", "0.26.5")
+  type ObservableExtensions[T] = IterableToObservable[T]
+
+  implicit class IterableToObservable[T](val iterable: Iterable[T]) extends AnyVal {
+    def toObservable: Observable[T] = Observable.from(iterable)
   }
 
   implicit class TryToObservable[T](val tryT: Try[T]) extends AnyVal {
@@ -38,4 +39,8 @@ package object scala {
   implicit class OptionToObservable[T](val opt: Option[T]) extends AnyVal {
     def toObservable: Observable[T] = Observable.from(opt)
   }
+
+  implicit class FutureToObservable[T](val future: Future[T]) extends AnyVal {
+    def toObservable(implicit ec: ExecutionContext): Observable[T] = Observable.from(future)
+  }
 }

src/test/scala/rx/lang/scala/ScalaTypesConversionsTests.scala

@@ -0,0 +1,118 @@
+package rx.lang.scala
+
+import org.junit.Test
+import org.scalatest.junit.JUnitSuite
+import rx.lang.scala.observers.TestSubscriber
+
+import scala.concurrent.Future
+import scala.util.{Failure, Success, Try}
+
+class ScalaTypesConversionsTests extends JUnitSuite {
+
+  @Test
+  def testIterableConversion() = {
+    val it = Seq("1", "2", "3")
+    val observer = TestSubscriber[String]()
+    it.toObservable.subscribe(observer)
+
+    observer.assertValues("1", "2", "3")
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testIterableEmptyConversion() = {
+    val it = List[String]()
+    val observer = TestSubscriber[String]()
+    it.toObservable.subscribe(observer)
+
+    observer.assertNoValues()
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testTrySuccessConversion() = {
+    val success = Success("abc")
+    val observer = TestSubscriber[String]()
+    success.toObservable.subscribe(observer)
+
+    observer.assertValue("abc")
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testTryFailureConversion() = {
+    val error = new IllegalArgumentException("test error")
+    val failure = Failure[String](error)
+    val observer = TestSubscriber[String]()
+    failure.toObservable.subscribe(observer)
+
+    observer.assertNoValues()
+    observer.assertError(error)
+    observer.assertNotCompleted()
+  }
+
+  @Test
+  def testOptionSomeConversion() = {
+    val some = Option("abc")
+    val observer = TestSubscriber[String]()
+    some.toObservable.subscribe(observer)
+
+    observer.assertValue("abc")
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testOptionNoneConversion() = {
+    val some = Option.empty[String]
+    val observer = TestSubscriber[String]()
+    some.toObservable.subscribe(observer)
+
+    observer.assertNoValues()
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testFutureSuccessfulConversion() = {
+    import scala.concurrent.ExecutionContext.Implicits.global
+    val fut = Future.successful("abc")
+    val observer = TestSubscriber[String]()
+    fut.toObservable.subscribe(observer)
+
+    observer.awaitTerminalEvent()
+    observer.assertValue("abc")
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testFutureSuccessfulConversion2() = {
+    import scala.concurrent.ExecutionContext.Implicits.global
+    val fut = Future { "abc" }
+    val observer = TestSubscriber[String]()
+    fut.toObservable.subscribe(observer)
+
+    observer.awaitTerminalEvent()
+    observer.assertValue("abc")
+    observer.assertNoErrors()
+    observer.assertCompleted()
+  }
+
+  @Test
+  def testFutureFailedConversion() = {
+    import scala.concurrent.ExecutionContext.Implicits.global
+    val error = new IllegalArgumentException("test error")
+    val fut = Future.failed[Unit](error)
+    val observer = TestSubscriber[Unit]()
+    fut.toObservable.subscribe(observer)
+
+    observer.awaitTerminalEvent()
+    observer.assertNoValues()
+    observer.assertError(error)
+    observer.assertNotCompleted()
+  }
+}