Add 'startWith' to RxScala

Author: zsxwing

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -756,4 +756,26 @@ class RxScalaDemo extends JUnitSuite {
       case e: IllegalArgumentException => println("IllegalArgumentException from skipWithException")
     }
   }
+
+  @Test def startWithExample1(): Unit = {
+    val o = List(2, 3).toObservable + 1
+    assertEquals(List(1, 2, 3), o.toBlockingObservable.toList)
+  }
+
+  @Test def startWithExample2(): Unit = {
+    val prepended = List(2, 4).toObservable
+    val o = List(5, 6, 7, 8).toObservable.filter(_ % 2 == 0).startWith(prepended)
+    assertEquals(List(2, 4, 6, 8), o.toBlockingObservable.toList)
+  }
+
+  @Test def startWithExample3(): Unit = {
+    val o = List(5, 6, 7, 8).toObservable.filter(_ % 2 == 0).startWith(List(2, 4))
+    assertEquals(List(2, 4, 6, 8), o.toBlockingObservable.toList)
+  }
+
+  @Test def startWithExample4(): Unit = {
+    val o = List(5, 6, 7, 8).toObservable.filter(_ % 2 == 0).startWith(Array(2, 4))
+    assertEquals(List(2, 4, 6, 8), o.toBlockingObservable.toList)
+  }
+
 }

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

@@ -77,6 +77,7 @@ import collection.JavaConversions._
  */
 trait Observable[+T]
 {
+  import scala.collection.JavaConverters._
   import scala.collection.Seq
   import scala.concurrent.duration.{Duration, TimeUnit}
   import rx.functions._
@@ -238,6 +239,65 @@ trait Observable[+T]
     toScalaObservable(rx.Observable.concat(o1, o2))
   }
 
+  /**
+   * Returns an Observable that emits a specified item before it begins to emit items emitted by the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/startWith.png">
+   *
+   * @param elem the item to emit
+   * @return an Observable that emits the specified item before it begins to emit items emitted by the source Observable
+   */
+  def +[U >: T](elem: U): Observable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable(thisJava.startWith(elem))
+  }
+
+  /**
+   * Returns an Observable that emits the items in a specified `Observable` before it begins to emit
+   * items emitted by the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/startWith.o.png">
+   *
+   * @param that an Observable that contains the items you want the modified Observable to emit first
+   * @return an Observable that emits the items in the specified `Observable` and then emits the items
+   *         emitted by the source Observable
+   */
+  def startWith[U >: T](that: Observable[U]): Observable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    val thatJava = that.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable(thisJava.startWith(thatJava))
+  }
+
+  /**
+   * Returns an Observable that emits the items in a specified `Iterable` before it begins to emit items
+   * emitted by the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/startWith.png">
+   *
+   * @param iterable an Iterable that contains the items you want the modified Observable to emit first
+   * @return an Observable that emits the items in the specified `Iterable` and then emits the items
+   *         emitted by the source Observable
+   */
+  def startWith[U >: T](iterable: Iterable[U]): Observable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable(thisJava.startWith(iterable.asJava))
+  }
+
+  /**
+   * Returns an Observable that emits the items in a specified `Iterable`, on a specified `Scheduler`, before it begins to emit items emitted by the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/startWith.s.png">
+   *
+   * @param iterable an Iterable that contains the items you want the modified Observable to emit first
+   * @param scheduler the Scheduler to emit the prepended values on
+   * @return an Observable that emits the items in the specified `Iterable`, on a specified `Scheduler`, and then emits the items
+   *         emitted by the source Observable
+   */
+  def startWith[U >: T](iterable: Iterable[U], scheduler: Scheduler): Observable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable(thisJava.startWith(iterable.asJava, scalaSchedulerToJavaScheduler(scheduler)))
+  }
+
   /**
    * Returns an Observable that emits the items emitted by several Observables, one after the
    * other.

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

@@ -100,7 +100,11 @@ class CompletenessTest extends JUnitSuite {
       "skipWhile(Func1[_ >: T, Boolean])" -> "dropWhile(T => Boolean)",
       "skipWhileWithIndex(Func2[_ >: T, Integer, Boolean])" -> unnecessary,
       "skipUntil(Observable[U])" -> "dropUntil(Observable[E])",
-      "startWith(Iterable[T])" -> "[unnecessary because we can just use `++` instead]",
+      "startWith(Array[T])" -> "startWith(Iterable[U])",
+      "startWith(Array[T], Scheduler)" -> "startWith(Iterable[U], Scheduler)",
+      "startWith(Iterable[T])" -> "startWith(Iterable[U])",
+      "startWith(Iterable[T], Scheduler)" -> "startWith(Iterable[U], Scheduler)",
+      "startWith(Observable[T])" -> "startWith(Observable[U])",
       "skipLast(Int)" -> "dropRight(Int)",
       "skipLast(Long, TimeUnit)" -> "dropRight(Duration)",
       "skipLast(Long, TimeUnit, Scheduler)" -> "dropRight(Duration, Scheduler)",
@@ -159,7 +163,7 @@ class CompletenessTest extends JUnitSuite {
       "zip(Iterable[_ <: Observable[_]], FuncN[_ <: R])" -> "[use `zip` in companion object and `map`]"
   ) ++ List.iterate("T", 9)(s => s + ", T").map(
       // all 9 overloads of startWith:
-      "startWith(" + _ + ")" -> "[unnecessary because we can just use `++` instead]"
+      "startWith(" + _ + ")" -> "[unnecessary because we can just use `+` instead]"
   ).toMap ++ List.iterate("Observable[_ <: T]", 9)(s => s + ", Observable[_ <: T]").map(
       // concat 2-9
       "concat(" + _ + ")" -> "[unnecessary because we can use `++` instead or `Observable(o1, o2, ...).concat`]"