Add retry variant to RxScala

Author: zsxwing

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

@@ -875,6 +875,28 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(List("alice", "bob", "carol"), o.retry(3).toBlockingObservable.toList)
   }
 
+  @Test def retryExample3(): Unit = {
+    var isFirst = true
+    val o = Observable {
+      (subscriber: Subscriber[String]) =>
+        if (isFirst) {
+          subscriber.onNext("alice")
+          subscriber.onError(new IOException("Oops"))
+          isFirst = false
+        }
+        else {
+          subscriber.onNext("bob")
+          subscriber.onError(new RuntimeException("Oops"))
+        }
+    }
+    o.retry {
+      (times, e) => e match {
+        case e: IOException => times <= 3
+        case _ => false
+      }
+    }.subscribe(s => println(s), e => e.printStackTrace())
+  }
+
   @Test def liftExample1(): Unit = {
     // Add "No. " in front of each item
     val o = List(1, 2, 3).toObservable.lift {

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

@@ -3086,6 +3086,22 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.retry())
   }
 
+  /**
+   * Returns an Observable that mirrors the source Observable, resubscribing to it if it calls `onError`
+   * and the predicate returns true for that specific exception and retry count.
+   *
+   * <img width="640" height="315" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/retry.png">
+   *
+   * @param predicate the predicate that determines if a resubscription may happen in case of a specific exception and retry count
+   * @return the source Observable modified with retry logic
+   */
+  def retry(predicate: (Int, Throwable) => Boolean): Observable[T] = {
+    val f = new Func2[java.lang.Integer, Throwable, java.lang.Boolean] {
+      def call(times: java.lang.Integer, e: Throwable): java.lang.Boolean = predicate(times, e)
+    }
+    toScalaObservable[T](asJavaObservable.retry(f))
+  }
+
   /**
    * Returns an Observable that repeats the sequence of items emitted by the source Observable indefinitely.
    * <p>

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

@@ -123,6 +123,7 @@ class CompletenessTest extends JUnitSuite {
       "replay(Func1[_ >: Observable[T], _ <: Observable[R]], Long, TimeUnit)" -> "replay(Observable[U] => Observable[R], Duration)",
       "replay(Func1[_ >: Observable[T], _ <: Observable[R]], Long, TimeUnit, Scheduler)" -> "replay(Observable[U] => Observable[R], Duration, Scheduler)",
       "replay(Func1[_ >: Observable[T], _ <: Observable[R]], Scheduler)" -> "replay(Observable[U] => Observable[R], Scheduler)",
+      "retry(Func2[Integer, Throwable, Boolean])" -> "retry((Int, Throwable) => Boolean)",
       "sample(Observable[U])" -> "sample(Observable[Any])",
       "scan(Func2[T, T, T])" -> unnecessary,
       "scan(R, Func2[R, _ >: T, R])" -> "scan(R)((R, T) => R)",