Merge pull request #1605 from jbripley/rxscala-retrywhensignature

RxScala: Update signatures of retryWhen/repeatWhen to match RxJava

Author: benjchristensen

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

@@ -1094,7 +1094,7 @@ class RxScalaDemo extends JUnitSuite {
     }).repeatWhen(attempts => {
       attempts.zipWith(Observable.from(1 to 3))((n, i) => i).flatMap(i => {
         println("delay repeat by " + i + " second(s)")
-        Observable.timer(Duration(i, TimeUnit.SECONDS)).materialize
+        Observable.timer(Duration(i, TimeUnit.SECONDS))
       })
     }, NewThreadScheduler()).toBlocking.foreach(s => println(s))
   }

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

@@ -3254,11 +3254,11 @@ trait Observable[+T]
    * @see <a href="https://github.com/Netflix/RxJava/wiki/Error-Handling-Operators#retrywhen">RxJava Wiki: retryWhen()</a>
    * @since 0.20
    */
-  def retryWhen(notificationHandler: Observable[Notification[Any]] => Observable[Notification[Any]], scheduler: Scheduler): Observable[T] = {
-    val f: Func1[_ >: rx.Observable[_ <: rx.Notification[_ <: Any]], _ <: rx.Observable[_ <: rx.Notification[_ <: Any]]] =
+  def retryWhen(notificationHandler: Observable[Notification[Any]] => Observable[Any], scheduler: Scheduler): Observable[T] = {
+    val f: Func1[_ >: rx.Observable[_ <: rx.Notification[_ <: Any]], _ <: rx.Observable[_ <: Any]] =
       (jOn: rx.Observable[_ <: rx.Notification[_ <: Any]]) => {
         val on = toScalaObservable[rx.Notification[_ <: Any]](jOn).map({ jN => toScalaNotification[Any](jN) })
-        notificationHandler(on).map({ n => n.asJavaNotification }).asJavaObservable
+        notificationHandler(on).asJavaObservable
       }
 
     toScalaObservable[T](asJavaObservable.retryWhen(f, scheduler))
@@ -3347,11 +3347,11 @@ trait Observable[+T]
    * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428.aspx">MSDN: Observable.Repeat</a>
    * @since 0.20
    */
-  def repeatWhen(notificationHandler: Observable[Notification[Any]] => Observable[Notification[Any]], scheduler: Scheduler): Observable[T] = {
-    val f: Func1[_ >: rx.Observable[_ <: rx.Notification[_ <: Any]], _ <: rx.Observable[_ <: rx.Notification[_ <: Any]]] =
+  def repeatWhen(notificationHandler: Observable[Notification[Any]] => Observable[Any], scheduler: Scheduler): Observable[T] = {
+    val f: Func1[_ >: rx.Observable[_ <: rx.Notification[_ <: Any]], _ <: rx.Observable[_ <: Any]] =
       (jOn: rx.Observable[_ <: rx.Notification[_ <: Any]]) => {
         val on = toScalaObservable[rx.Notification[_ <: Any]](jOn).map({ jN => toScalaNotification[Any](jN) })
-        notificationHandler(on).map({ n => n.asJavaNotification }).asJavaObservable
+        notificationHandler(on).asJavaObservable
       }
 
     toScalaObservable[T](asJavaObservable.repeatWhen(f, scheduler))
@@ -3377,11 +3377,11 @@ trait Observable[+T]
    * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428.aspx">MSDN: Observable.Repeat</a>
    * @since 0.20
    */
-  def repeatWhen(notificationHandler: Observable[Notification[Any]] => Observable[Notification[Any]]): Observable[T] = {
-    val f: Func1[_ >: rx.Observable[_ <: rx.Notification[_ <: Any]], _ <: rx.Observable[_ <: rx.Notification[_ <: Any]]] =
+  def repeatWhen(notificationHandler: Observable[Notification[Any]] => Observable[Any]): Observable[T] = {
+    val f: Func1[_ >: rx.Observable[_ <: rx.Notification[_ <: Any]], _ <: rx.Observable[_ <: Any]] =
       (jOn: rx.Observable[_ <: rx.Notification[_ <: Any]]) => {
         val on = toScalaObservable[rx.Notification[_ <: Any]](jOn).map({ jN => toScalaNotification[Any](jN) })
-        notificationHandler(on).map({ n => n.asJavaNotification }).asJavaObservable
+        notificationHandler(on).asJavaObservable
       }
 
     toScalaObservable[T](asJavaObservable.repeatWhen(f))