Merge branch 'master' of https://github.com/Netflix/RxJava into docs

Author: DavidMGross

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

@@ -1055,6 +1055,30 @@ class RxScalaDemo extends JUnitSuite {
     }.subscribe(s => println(s), e => e.printStackTrace())
   }
 
+  @Test def retryWhenExample(): Unit = {
+    Observable[String]({ subscriber =>
+      println("subscribing")
+      subscriber.onError(new RuntimeException("always fails"))
+    }).retryWhen(attempts => {
+      attempts.zipWith(Observable.from(1 to 3))((n, i) => i).flatMap(i => {
+        println("delay retry by " + i + " second(s)")
+        Observable.timer(Duration(i, TimeUnit.SECONDS))
+      })
+    }).toBlocking.foreach(s => println(s))
+  }
+
+  @Test def repeatWhenExample(): Unit = {
+    Observable[String]({ subscriber =>
+      println("subscribing")
+      subscriber.onCompleted()
+    }).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
+      })
+    }, NewThreadScheduler()).toBlocking.foreach(s => println(s))
+  }
+
   @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

@@ -3175,6 +3175,94 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.retry(f))
   }
 
+  /**
+   * Returns an Observable that emits the same values as the source observable with the exception of an
+   * {@code onError}. An {@code onError} notification from the source will result in the emission of a
+   * {@link Notification} to the Observable provided as an argument to the {@code notificationHandler}
+   * function. If the Observable returned {@code onCompletes} or {@code onErrors} then {@code retry} will call
+   * {@code onCompleted} or {@code onError} on the child subscription. Otherwise, this Observable will
+   * resubscribe to the source Observable.
+   * <p>
+   * <img width="640" height="430" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/retryWhen.f.png" alt="">
+   *
+   * Example:
+   *
+   * This retries 3 times, each time incrementing the number of seconds it waits.
+   *
+   * <pre> {@code
+   * Observable[String]({ subscriber =>
+   *   println("subscribing")
+   *   subscriber.onError(new RuntimeException("always fails"))
+   * }).retryWhen(attempts => {
+   *   attempts.zipWith(Observable.from(1 to 3))((n, i) => i).flatMap(i => {
+   *     println("delay retry by " + i + " second(s)")
+   *     Observable.timer(Duration(i, TimeUnit.SECONDS))
+   *   })
+   * }).toBlocking.foreach(s => println(s))
+   * } </pre>
+   *
+   * Output is:
+   *
+   * <pre> {@code
+   * subscribing
+   * delay retry by 1 second(s)
+   * subscribing
+   * delay retry by 2 second(s)
+   * subscribing
+   * delay retry by 3 second(s)
+   * subscribing
+   * } </pre>
+   * <dl>
+   *  <dt><b>Scheduler:</b></dt>
+   *  <dd>{@code retryWhen} operates by default on the {@code trampoline} {@link Scheduler}.</dd>
+   * </dl>
+   *
+   * @param notificationHandler receives an Observable of notifications with which a user can complete or error, aborting the
+   *            retry
+   * @return the source Observable modified with retry logic
+   * @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[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).asJavaObservable
+      }
+
+    toScalaObservable[T](asJavaObservable.retryWhen(f))
+  }
+
+  /**
+   * Returns an Observable that emits the same values as the source observable with the exception of an {@code onError}.
+   * An onError will emit a {@link Notification} to the observable provided as an argument to the notificationHandler
+   * func. If the observable returned {@code onCompletes} or {@code onErrors} then retry will call {@code onCompleted}
+   * or {@code onError} on the child subscription. Otherwise, this observable will resubscribe to the source observable, on a particular Scheduler.
+   * <p>
+   * <img width="640" height="430" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/retryWhen.f.png" alt="">
+   * <p>
+   * <dl>
+   *  <dt><b>Scheduler:</b></dt>
+   *  <dd>you specify which {@link Scheduler} this operator will use</dd>
+   * </dl>
+   *
+   * @param notificationHandler receives an Observable of notifications with which a user can complete or error, aborting the
+   *            retry
+   * @param scheduler the Scheduler on which to subscribe to the source Observable
+   * @return the source Observable modified with retry logic
+   * @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]]] =
+      (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
+      }
+
+    toScalaObservable[T](asJavaObservable.retryWhen(f, scheduler))
+  }
+
   /**
    * Returns an Observable that repeats the sequence of items emitted by the source Observable indefinitely.
    * <p>
@@ -3237,6 +3325,67 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.repeat(count, scheduler))
   }
 
+  /**
+   * Returns an Observable that emits the same values as the source Observable with the exception of an
+   * {@code onCompleted}. An {@code onCompleted} notification from the source will result in the emission of
+   * a {@link Notification} to the Observable provided as an argument to the {@code notificationHandler}
+   * function. If the Observable returned {@code onCompletes} or {@code onErrors} then {@code repeatWhen} will
+   * call {@code onCompleted} or {@code onError} on the child subscription. Otherwise, this Observable will
+   * resubscribe to the source Observable, on a particular Scheduler.
+   * <p>
+   * <img width="640" height="430" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/repeatWhen.f.png" alt="">
+   * <dl>
+   *  <dt><b>Scheduler:</b></dt>
+   *  <dd>you specify which {@link Scheduler} this operator will use</dd>
+   * </dl>
+   *
+   * @param notificationHandler receives an Observable of notifications with which a user can complete or error, aborting the repeat.
+   * @param scheduler the Scheduler to emit the items on
+   * @return the source Observable modified with repeat logic
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#repeatwhen">RxJava Wiki: repeatWhen()</a>
+   * @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]]] =
+      (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
+      }
+
+    toScalaObservable[T](asJavaObservable.repeatWhen(f, scheduler))
+  }
+
+  /**
+   * Returns an Observable that emits the same values as the source Observable with the exception of an
+   * {@code onCompleted}. An {@code onCompleted} notification from the source will result in the emission of
+   * a {@link Notification} to the Observable provided as an argument to the {@code notificationHandler}
+   * function. If the Observable returned {@code onCompletes} or {@code onErrors} then {@code repeatWhen} will
+   * call {@code onCompleted} or {@code onError} on the child subscription. Otherwise, this Observable will
+   * resubscribe to the source observable.
+   * <p>
+   * <img width="640" height="430" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/repeatWhen.f.png" alt="">
+   * <dl>
+   *  <dt><b>Scheduler:</b></dt>
+   *  <dd>{@code repeatWhen} operates by default on the {@code trampoline} {@link Scheduler}.</dd>
+   * </dl>
+   *
+   * @param notificationHandler receives an Observable of notifications with which a user can complete or error, aborting the repeat.
+   * @return the source Observable modified with repeat logic
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#repeatwhen">RxJava Wiki: repeatWhen()</a>
+   * @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]]] =
+      (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
+      }
+
+    toScalaObservable[T](asJavaObservable.repeatWhen(f))
+  }
+
   /**
    * Converts an Observable into a [[BlockingObservable]] (an Observable with blocking operators).
    *

rxjava-core/src/main/java/rx/internal/operators/BlockingOperatorMostRecent.java

@@ -16,6 +16,7 @@
 package rx.internal.operators;
 
 import java.util.Iterator;
+import java.util.NoSuchElementException;
 
 import rx.Observable;
 import rx.Subscriber;
@@ -41,54 +42,24 @@ public final class BlockingOperatorMostRecent {
      *         {@code initialValue} if {@code source} has not yet emitted any items
      */
     public static <T> Iterable<T> mostRecent(final Observable<? extends T> source, final T initialValue) {
-
         return new Iterable<T>() {
             @Override
             public Iterator<T> iterator() {
                 MostRecentObserver<T> mostRecentObserver = new MostRecentObserver<T>(initialValue);
-                final MostRecentIterator<T> nextIterator = new MostRecentIterator<T>(mostRecentObserver);
 
                 /**
                  * Subscribe instead of unsafeSubscribe since this is the final subscribe in the chain
                  * since it is for BlockingObservable.
                  */
                 source.subscribe(mostRecentObserver);
 
-                return nextIterator;
+                return mostRecentObserver.getIterable();
             }
         };
-
-    }
-
-    private static class MostRecentIterator<T> implements Iterator<T> {
-
-        private final MostRecentObserver<T> observer;
-
-        private MostRecentIterator(MostRecentObserver<T> observer) {
-            this.observer = observer;
-        }
-
-        @Override
-        public boolean hasNext() {
-            return !observer.isCompleted();
-        }
-
-        @Override
-        public T next() {
-            if (observer.getThrowable() != null) {
-                throw Exceptions.propagate(observer.getThrowable());
-            }
-            return observer.getRecentValue();
-        }
-
-        @Override
-        public void remove() {
-            throw new UnsupportedOperationException("Read only iterator");
-        }
     }
 
     private static class MostRecentObserver<T> extends Subscriber<T> {
-        static final NotificationLite<Object> nl = NotificationLite.instance();
+        final NotificationLite<T> nl = NotificationLite.instance();
         volatile Object value;
 
         private MostRecentObserver(T value) {
@@ -110,19 +81,47 @@ public void onNext(T args) {
             value = nl.next(args);
         }
 
-        private boolean isCompleted() {
-            return nl.isCompleted(value);
-        }
-
-        private Throwable getThrowable() {
-            Object v = value;
-            return nl.isError(v) ? nl.getError(v) : null;
-        }
-
-        @SuppressWarnings("unchecked")
-        private T getRecentValue() {
-            return (T)value;
+        /**
+         * The {@link Iterator} return is not thread safe. In other words don't call {@link Iterator#hasNext()} in one
+         * thread expect {@link Iterator#next()} called from a different thread to work.
+         * @return
+         */
+        public Iterator<T> getIterable() {
+            return new Iterator<T>() {
+                /**
+                 * buffer to make sure that the state of the iterator doesn't change between calling hasNext() and next().
+                 */
+                private Object buf = null;
+
+                @Override
+                public boolean hasNext() {
+                    buf = value;
+                    return !nl.isCompleted(buf);
+                }
+
+                @Override
+                public T next() {
+                    try {
+                        // if hasNext wasn't called before calling next.
+                        if (buf == null)
+                            buf = value;
+                        if (nl.isCompleted(buf))
+                            throw new NoSuchElementException();
+                        if (nl.isError(buf)) {
+                            throw Exceptions.propagate(nl.getError(buf));
+                        }
+                        return nl.getValue(buf);
+                    }
+                    finally {
+                        buf = null;
+                    }
+                }
+
+                @Override
+                public void remove() {
+                    throw new UnsupportedOperationException("Read only iterator");
+                }
+            };
         }
-
     }
 }

rxjava-core/src/test/java/rx/internal/operators/BlockingOperatorMostRecentTest.java

@@ -34,6 +34,10 @@
 import rx.subjects.Subject;
 
 public class BlockingOperatorMostRecentTest {
+    @Test
+    public void testMostRecentNull() {
+        assertEquals(null, Observable.<Void>never().toBlocking().mostRecent(null).iterator().next());
+    }
 
     @Test
     public void testMostRecent() {