Add delay variants to RxScala

Author: zsxwing

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

@@ -642,6 +642,21 @@ class RxScalaDemo extends JUnitSuite {
     println(result)
   }
 
+  @Test def delayExample3(): Unit = {
+    val o = List(100, 500, 200).toObservable.delay(
+      (i: Int) => Observable.items(i).delay(i millis)
+    )
+    o.toBlockingObservable.foreach(println(_))
+  }
+
+  @Test def delayExample4(): Unit = {
+    val o = List(100, 500, 200).toObservable.delay(
+      () => Observable.interval(500 millis).take(1),
+      (i: Int) => Observable.items(i).delay(i millis)
+    )
+    o.toBlockingObservable.foreach(println(_))
+  }
+
   @Test def delaySubscriptionExample(): Unit = {
     val o = List(100L, 200L, 300L).toObservable.delaySubscription(2 seconds)
     val result = o.toBlockingObservable.toList

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

@@ -2690,6 +2690,57 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.delay(delay.length, delay.unit, scheduler))
   }
 
+  /**
+   * Returns an Observable that delays the emissions of the source Observable via another Observable on a
+   * per-item basis.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/delay.o.png">
+   * <p>
+   * Note: the resulting Observable will immediately propagate any `onError` notification
+   * from the source Observable.
+   *
+   * @param itemDelay a function that returns an Observable for each item emitted by the source Observable, which is
+   *                  then used to delay the emission of that item by the resulting Observable until the Observable
+   *                  returned from `itemDelay` emits an item
+   * @return an Observable that delays the emissions of the source Observable via another Observable on a per-item basis
+   */
+  def delay(itemDelay: T => Observable[Any]): Observable[T] = {
+    val itemDelayJava = new Func1[T, rx.Observable[Any]] {
+      override def call(t: T): rx.Observable[Any] =
+        itemDelay(t).asJavaObservable.asInstanceOf[rx.Observable[Any]]
+    }
+    toScalaObservable[T](asJavaObservable.delay[Any](itemDelayJava))
+  }
+
+  /**
+   * Returns an Observable that delays the subscription to and emissions from the souce Observable via another
+   * Observable on a per-item basis.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/delay.oo.png">
+   * <p>
+   * Note: the resulting Observable will immediately propagate any `onError` notification
+   * from the source Observable.
+   *
+   * @param subscriptionDelay a function that returns an Observable that triggers the subscription to the source Observable
+   *                          once it emits any item
+   * @param itemDelay a function that returns an Observable for each item emitted by the source Observable, which is
+   *                  then used to delay the emission of that item by the resulting Observable until the Observable
+   *                  returned from `itemDelay` emits an item
+   * @return an Observable that delays the subscription and emissions of the source Observable via another
+   *         Observable on a per-item basis
+   */
+  def delay(subscriptionDelay: () => Observable[Any], itemDelay: T => Observable[Any]): Observable[T] = {
+    val subscriptionDelayJava = new Func0[rx.Observable[Any]] {
+      override def call(): rx.Observable[Any] =
+        subscriptionDelay().asJavaObservable.asInstanceOf[rx.Observable[Any]]
+    }
+    val itemDelayJava = new Func1[T, rx.Observable[Any]] {
+      override def call(t: T): rx.Observable[Any] =
+        itemDelay(t).asJavaObservable.asInstanceOf[rx.Observable[Any]]
+    }
+    toScalaObservable[T](asJavaObservable.delay[Any, Any](subscriptionDelayJava, itemDelayJava))
+  }
+
   /**
    * Return an Observable that delays the subscription to the source Observable by a given amount of time.
    *