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

Author: samuelgruetter

rxjava-core/src/main/java/rx/Observable.java

@@ -52,6 +52,7 @@
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
+import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
 import rx.operators.OperationScan;
 import rx.operators.OperationSkip;
@@ -1891,29 +1892,6 @@ public static <T> Observable<T> from(Future<? extends T> future, long timeout, T
         return create(OperationToObservableFuture.toObservableFuture(future, timeout, unit));
     }
 
-    /**
-     * <p>
-     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
-     * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
-     * new Observable will be the result of the function applied to the first item emitted by {@code w0} and the first item emitted by {@code w1}; the second item emitted by
-     * the new Observable will be the result of the function applied to the second item emitted by {@code w0} and the second item emitted by {@code w1}; and so forth.
-     * <p>
-     * The resulting {@code Observable<R>} returned from {@code zip} will invoke {@link Observer#onNext onNext} as many times as the number of {@code onNext} invocations
-     * of the source Observable that emits the fewest items.
-     * 
-     * @param o1
-     *            one source Observable
-     * @param o2
-     *            another source Observable
-     * @param zipFunction
-     *            a function that, when applied to an item emitted by each of the source
-     *            Observables, results in an item that will be emitted by the resulting Observable
-     * @return an Observable that emits the zipped results
-     */
-    public static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, zipFunction));
-    }
-
     /**
      * Returns an Observable that emits Boolean values that indicate whether the pairs of items
      * emitted by two source Observables are equal.
@@ -1960,6 +1938,31 @@ public static <T> Observable<Boolean> sequenceEqual(Observable<? extends T> firs
         return zip(first, second, equality);
     }
 
+    /**
+     * Returns an Observable that emits the results of a function of your choosing applied to
+     * combinations of two items emitted, in sequence, by two other Observables.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
+     * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
+     * new Observable will be the result of the function applied to the first item emitted by {@code w0} and the first item emitted by {@code w1}; the second item emitted by
+     * the new Observable will be the result of the function applied to the second item emitted by {@code w0} and the second item emitted by {@code w1}; and so forth.
+     * <p>
+     * The resulting {@code Observable<R>} returned from {@code zip} will invoke {@link Observer#onNext onNext} as many times as the number of {@code onNext} invocations
+     * of the source Observable that emits the fewest items.
+     * 
+     * @param o1
+     *            one source Observable
+     * @param o2
+     *            another source Observable
+     * @param zipFunction
+     *            a function that, when applied to an item emitted by each of the source
+     *            Observables, results in an item that will be emitted by the resulting Observable
+     * @return an Observable that emits the zipped results
+     */
+    public static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> zipFunction) {
+        return create(OperationZip.zip(o1, o2, zipFunction));
+    }
+
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
      * combinations of three items emitted, in sequence, by three other Observables.
@@ -2021,7 +2024,7 @@ public static <T1, T2, T3, T4, R> Observable<R> zip(Observable<? extends T1> o1,
 
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
-     * combinations of four items emitted, in sequence, by four other Observables.
+     * combinations of five items emitted, in sequence, by five other Observables.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
@@ -2054,7 +2057,7 @@ public static <T1, T2, T3, T4, T5, R> Observable<R> zip(Observable<? extends T1>
 
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
-     * combinations of four items emitted, in sequence, by four other Observables.
+     * combinations of six items emitted, in sequence, by six other Observables.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
@@ -2090,7 +2093,7 @@ public static <T1, T2, T3, T4, T5, T6, R> Observable<R> zip(Observable<? extends
 
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
-     * combinations of four items emitted, in sequence, by four other Observables.
+     * combinations of seven items emitted, in sequence, by seven other Observables.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
@@ -2128,7 +2131,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, R> Observable<R> zip(Observable<? ext
 
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
-     * combinations of four items emitted, in sequence, by four other Observables.
+     * combinations of eight items emitted, in sequence, by eight other Observables.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
@@ -2168,7 +2171,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Observable<R> zip(Observable<?
 
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
-     * combinations of four items emitted, in sequence, by four other Observables.
+     * combinations of nine items emitted, in sequence, by nine other Observables.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
@@ -2680,7 +2683,8 @@ public Observable<Observable<T>> window(long timespan, long timeshift, TimeUnit
 
     /**
      * Returns an Observable that emits the results of a function of your choosing applied to
-     * combinations of four items emitted, in sequence, by four other Observables.
+     * combinations of N items emitted, in sequence, by N other Observables as provided by an Iterable.
+     * 
      * <p> {@code zip} applies this function in strict sequence, so the first item emitted by the
      * new Observable will be the result of the function applied to the first item emitted by
      * all of the Observalbes; the second item emitted by the new Observable will be the result of
@@ -2727,7 +2731,7 @@ public Observable<R> call(List<? extends Observable<?>> wsList) {
      *            Observables, results in an item that will be emitted by the resulting Observable
      * @return an Observable that emits the zipped results
      */
-    public static <R> Observable<R> zip(Collection<? extends Observable<?>> ws, FuncN<? extends R> zipFunction) {
+    public static <R> Observable<R> zip(Iterable<? extends Observable<?>> ws, FuncN<? extends R> zipFunction) {
         return create(OperationZip.zip(ws, zipFunction));
     }
 
@@ -3125,6 +3129,42 @@ public static Observable<Double> averageDoubles(Observable<Double> source) {
     public ConnectableObservable<T> replay() {
         return OperationMulticast.multicast(this, ReplaySubject.<T> create());
     }
+    
+    /**
+     * Retry subscription to origin Observable upto given retry count.
+     * <p>
+     * If {@link Observer#onError} is invoked the source Observable will be re-subscribed to as many times as defined by retryCount.
+     * <p>
+     * Any {@link Observer#onNext} calls received on each attempt will be emitted and concatenated together.
+     * <p>
+     * For example, if an Observable fails on first time but emits [1, 2] then succeeds the second time and
+     * emits [1, 2, 3, 4, 5] then the complete output would be [1, 2, 1, 2, 3, 4, 5, onCompleted].
+     * 
+     * @param retryCount
+     *            Number of retry attempts before failing.
+     * @return Observable with retry logic.
+     */
+    public Observable<T> retry(int retryCount) {
+        return create(OperationRetry.retry(this, retryCount));
+    }
+
+    /**
+     * Retry subscription to origin Observable whenever onError is called (infinite retry count).
+     * <p>
+     * If {@link Observer#onError} is invoked the source Observable will be re-subscribed to.
+     * <p>
+     * Any {@link Observer#onNext} calls received on each attempt will be emitted and concatenated together.
+     * <p>
+     * For example, if an Observable fails on first time but emits [1, 2] then succeeds the second time and
+     * emits [1, 2, 3, 4, 5] then the complete output would be [1, 2, 1, 2, 3, 4, 5, onCompleted].
+     * 
+     * @param retryCount
+     *            Number of retry attempts before failing.
+     * @return Observable with retry logic.
+     */
+    public Observable<T> retry() {
+        return create(OperationRetry.retry(this));
+    }
 
     /**
      * This method has similar behavior to {@link #replay} except that this auto-subscribes to

rxjava-core/src/main/java/rx/operators/OperationRetry.java

@@ -0,0 +1,203 @@
+package rx.operators;
+
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+import org.mockito.InOrder;
+
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.concurrency.Schedulers;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.Subscriptions;
+import rx.util.functions.Action0;
+
+public class OperationRetry {
+
+    private static final int INFINITE_RETRY = -1;
+
+    public static <T> OnSubscribeFunc<T> retry(final Observable<T> observable, final int retryCount) {
+        return new Retry<T>(observable, retryCount);
+    }
+
+    public static <T> OnSubscribeFunc<T> retry(final Observable<T> observable) {
+        return new Retry<T>(observable, INFINITE_RETRY);
+    }
+
+    private static class Retry<T> implements OnSubscribeFunc<T> {
+
+        private final Observable<T> source;
+        private final int retryCount;
+        private final AtomicInteger attempts = new AtomicInteger(0);
+        private final CompositeSubscription subscription = new CompositeSubscription();
+
+        public Retry(Observable<T> source, int retryCount) {
+            this.source = source;
+            this.retryCount = retryCount;
+        }
+
+        @Override
+        public Subscription onSubscribe(Observer<? super T> observer) {
+            subscription.add(Schedulers.currentThread().schedule(attemptSubscription(observer)));
+            return subscription;
+        }
+
+        private Action0 attemptSubscription(final Observer<? super T> observer) {
+            return new Action0() {
+
+                @Override
+                public void call() {
+                    attempts.incrementAndGet();
+                    source.subscribe(new Observer<T>() {
+
+                        @Override
+                        public void onCompleted() {
+                            observer.onCompleted();
+                        }
+
+                        @Override
+                        public void onError(Throwable e) {
+                            if ((retryCount == INFINITE_RETRY || attempts.get() <= retryCount) && !subscription.isUnsubscribed()) {
+                                // retry again
+                                // remove the last subscription since we have completed (so as we retry we don't build up a huge list)
+                                subscription.removeLast();
+                                // add the new subscription and schedule a retry
+                                subscription.add(Schedulers.currentThread().schedule(attemptSubscription(observer)));
+                            } else {
+                                // give up and pass the failure
+                                observer.onError(e);
+                            }
+                        }
+
+                        @Override
+                        public void onNext(T v) {
+                            observer.onNext(v);
+                        }
+                    });
+
+                }
+            };
+        }
+
+    }
+
+    public static class UnitTest {
+
+        @Test
+        public void testOriginFails() {
+            @SuppressWarnings("unchecked")
+            Observer<String> observer = mock(Observer.class);
+            Observable<String> origin = Observable.create(new FuncWithErrors(2));
+            origin.subscribe(observer);
+
+            InOrder inOrder = inOrder(observer);
+            inOrder.verify(observer, times(1)).onNext("beginningEveryTime");
+            inOrder.verify(observer, times(1)).onError(any(RuntimeException.class));
+            inOrder.verify(observer, never()).onNext("onSuccessOnly");
+            inOrder.verify(observer, never()).onCompleted();
+        }
+
+        @Test
+        public void testRetryFail() {
+            int NUM_RETRIES = 1;
+            int NUM_FAILURES = 2;
+            @SuppressWarnings("unchecked")
+            Observer<String> observer = mock(Observer.class);
+            Observable<String> origin = Observable.create(new FuncWithErrors(NUM_FAILURES));
+            Observable.create(retry(origin, NUM_RETRIES)).subscribe(observer);
+
+            InOrder inOrder = inOrder(observer);
+            // should show 2 attempts (first time fail, second time (1st retry) fail)
+            inOrder.verify(observer, times(1 + NUM_RETRIES)).onNext("beginningEveryTime");
+            // should only retry once, fail again and emit onError
+            inOrder.verify(observer, times(1)).onError(any(RuntimeException.class));
+            // no success
+            inOrder.verify(observer, never()).onNext("onSuccessOnly");
+            inOrder.verify(observer, never()).onCompleted();
+            inOrder.verifyNoMoreInteractions();
+        }
+
+        @Test
+        public void testRetrySuccess() {
+            int NUM_RETRIES = 3;
+            int NUM_FAILURES = 2;
+            @SuppressWarnings("unchecked")
+            Observer<String> observer = mock(Observer.class);
+            Observable<String> origin = Observable.create(new FuncWithErrors(NUM_FAILURES));
+            Observable.create(retry(origin, NUM_RETRIES)).subscribe(observer);
+
+            InOrder inOrder = inOrder(observer);
+            // should show 3 attempts
+            inOrder.verify(observer, times(1 + NUM_FAILURES)).onNext("beginningEveryTime");
+            // should have no errors
+            inOrder.verify(observer, never()).onError(any(Throwable.class));
+            // should have a single success
+            inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
+            // should have a single successful onCompleted
+            inOrder.verify(observer, times(1)).onCompleted();
+            inOrder.verifyNoMoreInteractions();
+        }
+        
+        @Test
+        public void testInfiniteRetry() {
+            int NUM_FAILURES = 20;
+            @SuppressWarnings("unchecked")
+            Observer<String> observer = mock(Observer.class);
+            Observable<String> origin = Observable.create(new FuncWithErrors(NUM_FAILURES));
+            Observable.create(retry(origin)).subscribe(observer);
+
+            InOrder inOrder = inOrder(observer);
+            // should show 3 attempts
+            inOrder.verify(observer, times(1 + NUM_FAILURES)).onNext("beginningEveryTime");
+            // should have no errors
+            inOrder.verify(observer, never()).onError(any(Throwable.class));
+            // should have a single success
+            inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
+            // should have a single successful onCompleted
+            inOrder.verify(observer, times(1)).onCompleted();
+            inOrder.verifyNoMoreInteractions();
+        }
+
+        public static class FuncWithErrors implements OnSubscribeFunc<String> {
+
+            private final int numFailures;
+            private final AtomicInteger count = new AtomicInteger(0);
+
+            FuncWithErrors(int count) {
+                this.numFailures = count;
+            }
+
+            @Override
+            public Subscription onSubscribe(Observer<? super String> o) {
+                o.onNext("beginningEveryTime");
+                if (count.incrementAndGet() <= numFailures) {
+                    o.onError(new RuntimeException("forced failure: " + count.get()));
+                } else {
+                    o.onNext("onSuccessOnly");
+                    o.onCompleted();
+                }
+                return Subscriptions.empty();
+            }
+        };
+    }
+}

rxjava-core/src/main/java/rx/operators/OperationZip.java

@@ -147,7 +147,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> OnSubscribeFunc<R> zip(Obs
         return a;
     }
 
-    public static <R> OnSubscribeFunc<R> zip(Collection<? extends Observable<?>> ws, FuncN<? extends R> zipFunction) {
+    public static <R> OnSubscribeFunc<R> zip(Iterable<? extends Observable<?>> ws, FuncN<? extends R> zipFunction) {
         Aggregator<R> a = new Aggregator<R>(zipFunction);
         for (Observable<?> w : ws) {
             ZipObserver<R, Object> zipObserver = new ZipObserver<R, Object>(a, w);