Add 'dropUntil' to RxScala

Author: zsxwing

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

@@ -426,6 +426,15 @@ class RxScalaDemo extends JUnitSuite {
     )
   }
 
+  @Test def dropUntilExample() {
+    val o = List("Alice", "Bob", "Carlos").toObservable.zip(
+      Observable.interval(700 millis, IOScheduler())).map(_._1) // emit every 700 millis
+    val other = List(1).toObservable.delay(1 seconds)
+    println(
+      o.dropUntil(other).toBlockingObservable.toList // output List("Bob", "Carlos")
+    )
+  }
+
   def square(x: Int): Int = {
     println(s"$x*$x is being calculated on thread ${Thread.currentThread().getId}")
     Thread.sleep(100) // calculating a square is heavy work :)

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

@@ -1327,6 +1327,22 @@ trait Observable[+T]
     toScalaObservable(asJavaObservable.skipLast(time.length, time.unit, scheduler))
   }
 
+  /**
+   * Returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/skipUntil.png">
+   *
+   * @param other the second Observable that has to emit an item before the source Observable's elements begin
+   *              to be mirrored by the resulting Observable
+   * @return an Observable that skips items from the source Observable until the second Observable emits an
+   *         item, then emits the remaining items
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-skipuntil">RxJava Wiki: skipUntil()</a>
+   * @see <a href="http://msdn.microsoft.com/en-us/library/hh229358.aspx">MSDN: Observable.SkipUntil</a>
+   */
+  def dropUntil[E](other: Observable[E]): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.skipUntil(other))
+  }
+
   /**
    * Returns an Observable that emits only the first `num` items emitted by the source
    * Observable.

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

@@ -98,6 +98,7 @@ class CompletenessTest extends JUnitSuite {
       "skip(Long, TimeUnit, Scheduler)" -> "drop(Duration, Scheduler)",
       "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]",
       "skipLast(Int)" -> "dropRight(Int)",
       "skipLast(Long, TimeUnit)" -> "dropRight(Duration)",