Add debounce variant to RxScala

Author: zsxwing

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

@@ -1070,4 +1070,17 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(10, o2.size.toBlockingObservable.single)
     assertEquals(1000, o2.flatten.size.toBlockingObservable.single)
   }
+
+  @Test def debounceExample() {
+    val o = Observable.interval(100 millis).take(20).debounce {
+      n =>
+        if (n % 2 == 0) {
+          Observable.interval(50 millis)
+        }
+        else {
+          Observable.interval(150 millis)
+        }
+    }
+    o.toBlockingObservable.foreach(println(_))
+  }
 }

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

@@ -2299,6 +2299,23 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.throttleWithTimeout(timeout.length, timeout.unit))
   }
 
+  /**
+   * Return an Observable that mirrors the source Observable, except that it drops items emitted by the source
+   * Observable that are followed by another item within a computed debounce duration.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/debounce.f.png">
+   *
+   * @param debounceSelector function to retrieve a sequence that indicates the throttle duration for each item
+   * @return an Observable that omits items emitted by the source Observable that are followed by another item
+   *         within a computed debounce duration
+   */
+  def debounce(debounceSelector: T => Observable[Any]): Observable[T] = {
+    val fJava = new rx.functions.Func1[T, rx.Observable[Any]] {
+      override def call(t: T) = debounceSelector(t).asJavaObservable.asInstanceOf[rx.Observable[Any]]
+    }
+    toScalaObservable[T](asJavaObservable.debounce[Any](fJava))
+  }
+
   /**
    * Debounces by dropping all values that are followed by newer values before the timeout value expires. The timer resets on each `onNext` call.
    *