Add onErrorFlatMap to RxScala

Author: zsxwing

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

@@ -1176,4 +1176,38 @@ class RxScalaDemo extends JUnitSuite {
       .take(20)
       .toBlocking.foreach(println)
   }
+
+  @Test def onErrorResumeNextExample() {
+    val o = Observable {
+      (subscriber: Subscriber[Int]) =>
+        subscriber.onNext(1)
+        subscriber.onNext(2)
+        subscriber.onError(new IOException("Oops"))
+        subscriber.onNext(3)
+        subscriber.onNext(4)
+    }
+    o.onErrorResumeNext(_ => Observable.items(10, 11, 12)).subscribe(println(_))
+  }
+
+  @Test def onErrorFlatMapExample() {
+    val o = Observable {
+      (subscriber: Subscriber[Int]) =>
+        subscriber.onNext(1)
+        subscriber.onNext(2)
+        subscriber.onError(new IOException("Oops"))
+        subscriber.onNext(3)
+        subscriber.onNext(4)
+    }
+    o.onErrorFlatMap((_, _) => Observable.items(10, 11, 12)).subscribe(println(_))
+  }
+
+  @Test def onErrorFlatMapExample2() {
+    val o = Observable.items(4, 2, 0).map(16 / _).onErrorFlatMap {
+      (e, op) => op match {
+        case Some(v) if v == 0 => Observable.items(Int.MinValue)
+        case _ => Observable.empty
+      }
+    }
+    o.subscribe(println(_))
+  }
 }

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

@@ -1208,6 +1208,30 @@ trait Observable[+T]
     toScalaObservable[U](asJavaObservable.onErrorReturn(f2))
   }
 
+  /**
+   * Intercepts `onError` notifications from the source Observable and replaces them with the
+   * `onNext` emissions of an Observable returned by a specified function. This allows the source
+   * sequence to continue even if it issues multiple `onError` notifications.
+   *
+   * <img width="640" height="310" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/onErrorFlatMap.png">
+   *
+   * @param resumeFunction a function that accepts an `Throwable` and an `Option` associated with this error representing
+   *                       the Throwable issued by the source Observable, and returns an Observable that emits items
+   *                       that will be emitted in place of the error. If no value is associated with the error, the value
+   *                       will be `None`.
+   * @return the original Observable, with appropriately modified behavior
+   */
+  def onErrorFlatMap[U >: T](resumeFunction: (Throwable, Option[Any]) => Observable[U]): Observable[U] = {
+    val f = new Func1[rx.exceptions.OnErrorThrowable, rx.Observable[_ <: U]] {
+      override def call(t: rx.exceptions.OnErrorThrowable): rx.Observable[_ <: U] = {
+        val v = if (t.isValueNull) Some(t.getValue) else None
+        resumeFunction(t.getCause, v).asJavaObservable
+      }
+    }
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable[U](thisJava.onErrorFlatMap(f))
+  }
+
   /**
    * Returns an Observable that applies a function of your choosing to the first item emitted by a
    * source Observable, then feeds the result of that function along with the second item emitted

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

@@ -106,6 +106,7 @@ class CompletenessTest extends JUnitSuite {
       "onErrorResumeNext(Observable[_ <: T])" -> "onErrorResumeNext(Observable[U])",
       "onErrorReturn(Func1[Throwable, _ <: T])" -> "onErrorReturn(Throwable => U)",
       "onExceptionResumeNext(Observable[_ <: T])" -> "onExceptionResumeNext(Observable[U])",
+      "onErrorFlatMap(Func1[OnErrorThrowable, _ <: Observable[_ <: T]])" -> "onErrorFlatMap((Throwable, Option[Any]) => Observable[U])",
       "parallel(Func1[Observable[T], Observable[R]])" -> "parallel(Observable[T] => Observable[R])",
       "parallel(Func1[Observable[T], Observable[R]], Scheduler)" -> "parallel(Observable[T] => Observable[R], Scheduler)",
       "publish(T)" -> "publish(U)",