Fix retry with predicate ignoring backpressure.

Author: akarnokd

src/main/java/rx/Observable.java

@@ -6576,6 +6576,8 @@ public final Observable<T> retry(final long count) {
      * <p>
      * <img width="640" height="315" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/retry.png" alt="">
      * <dl>
+     *  <dt><b>Backpressure Support:</b></dt>
+     *  <dd>This operator honors backpressure.</td>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code retry} operates by default on the {@code trampoline} {@link Scheduler}.</dd>
      * </dl>

src/main/java/rx/internal/operators/OperatorRetryWithPredicate.java

@@ -16,11 +16,14 @@
 package rx.internal.operators;
 
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
 import rx.Observable;
+import rx.Producer;
 import rx.Scheduler;
 import rx.Subscriber;
 import rx.functions.Action0;
 import rx.functions.Func2;
+import rx.internal.producers.ProducerArbiter;
 import rx.schedulers.Schedulers;
 import rx.subscriptions.SerialSubscription;
 
@@ -38,88 +41,99 @@ public Subscriber<? super Observable<T>> call(final Subscriber<? super T> child)
         final SerialSubscription serialSubscription = new SerialSubscription();
         // add serialSubscription so it gets unsubscribed if child is unsubscribed
         child.add(serialSubscription);
-        
-        return new SourceSubscriber<T>(child, predicate, inner, serialSubscription);
+        ProducerArbiter pa = new ProducerArbiter();
+        child.setProducer(pa);
+        return new SourceSubscriber<T>(child, predicate, inner, serialSubscription, pa);
     }
 
     static final class SourceSubscriber<T> extends Subscriber<Observable<T>> {
         final Subscriber<? super T> child;
         final Func2<Integer, Throwable, Boolean> predicate;
         final Scheduler.Worker inner;
         final SerialSubscription serialSubscription;
+        final ProducerArbiter pa;
 
         volatile int attempts;
         @SuppressWarnings("rawtypes")
         static final AtomicIntegerFieldUpdater<SourceSubscriber> ATTEMPTS_UPDATER
                 = AtomicIntegerFieldUpdater.newUpdater(SourceSubscriber.class, "attempts");
 
-        public SourceSubscriber(Subscriber<? super T> child, final Func2<Integer, Throwable, Boolean> predicate, Scheduler.Worker inner, 
-                SerialSubscription serialSubscription) {
+        public SourceSubscriber(Subscriber<? super T> child, 
+                final Func2<Integer, Throwable, Boolean> predicate, 
+                Scheduler.Worker inner, 
+                SerialSubscription serialSubscription,
+                ProducerArbiter pa) {
             this.child = child;
             this.predicate = predicate;
             this.inner = inner;
             this.serialSubscription = serialSubscription;
+            this.pa = pa;
         }
 
 
         @Override
-            public void onCompleted() {
-                // ignore as we expect a single nested Observable<T>
-            }
+        public void onCompleted() {
+            // ignore as we expect a single nested Observable<T>
+        }
 
-            @Override
-            public void onError(Throwable e) {
-                child.onError(e);
-            }
+        @Override
+        public void onError(Throwable e) {
+            child.onError(e);
+        }
 
-            @Override
-            public void onNext(final Observable<T> o) {
-                inner.schedule(new Action0() {
+        @Override
+        public void onNext(final Observable<T> o) {
+            inner.schedule(new Action0() {
 
-                    @Override
-                    public void call() {
-                        final Action0 _self = this;
-                        ATTEMPTS_UPDATER.incrementAndGet(SourceSubscriber.this);
+                @Override
+                public void call() {
+                    final Action0 _self = this;
+                    ATTEMPTS_UPDATER.incrementAndGet(SourceSubscriber.this);
 
-                        // new subscription each time so if it unsubscribes itself it does not prevent retries
-                        // by unsubscribing the child subscription
-                        Subscriber<T> subscriber = new Subscriber<T>() {
-                            boolean done;
-                            @Override
-                            public void onCompleted() {
-                                if (!done) {
-                                    done = true;
-                                    child.onCompleted();
-                                }
+                    // new subscription each time so if it unsubscribes itself it does not prevent retries
+                    // by unsubscribing the child subscription
+                    Subscriber<T> subscriber = new Subscriber<T>() {
+                        boolean done;
+                        @Override
+                        public void onCompleted() {
+                            if (!done) {
+                                done = true;
+                                child.onCompleted();
                             }
+                        }
 
-                            @Override
-                            public void onError(Throwable e) {
-                                if (!done) {
-                                    done = true;
-                                    if (predicate.call(attempts, e) && !inner.isUnsubscribed()) {
-                                        // retry again
-                                        inner.schedule(_self);
-                                    } else {
-                                        // give up and pass the failure
-                                        child.onError(e);
-                                    }
+                        @Override
+                        public void onError(Throwable e) {
+                            if (!done) {
+                                done = true;
+                                if (predicate.call(attempts, e) && !inner.isUnsubscribed()) {
+                                    // retry again
+                                    inner.schedule(_self);
+                                } else {
+                                    // give up and pass the failure
+                                    child.onError(e);
                                 }
                             }
+                        }
 
-                            @Override
-                            public void onNext(T v) {
-                                if (!done) {
-                                    child.onNext(v);
-                                }
+                        @Override
+                        public void onNext(T v) {
+                            if (!done) {
+                                child.onNext(v);
+                                pa.produced(1);
                             }
+                        }
 
-                        };
-                        // register this Subscription (and unsubscribe previous if exists) 
-                        serialSubscription.set(subscriber);
-                        o.unsafeSubscribe(subscriber);
-                    }
-                });
-            }
+                        @Override
+                        public void setProducer(Producer p) {
+                            pa.setProducer(p);
+                        }
+                    };
+                    // register this Subscription (and unsubscribe previous if exists) 
+                    serialSubscription.set(subscriber);
+                    o.unsafeSubscribe(subscriber);
+                }
+            });
+        }
     }
 }

src/test/java/rx/internal/operators/OperatorRetryWithPredicateTest.java

@@ -20,20 +20,27 @@
 import static org.mockito.Mockito.*;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 import org.mockito.InOrder;
 
-import rx.*;
+import rx.Observable;
 import rx.Observable.OnSubscribe;
+import rx.Observer;
+import rx.Subscriber;
+import rx.Subscription;
 import rx.exceptions.TestException;
-import rx.functions.*;
+import rx.functions.Action1;
+import rx.functions.Func1;
+import rx.functions.Func2;
 import rx.observers.TestSubscriber;
 import rx.subjects.PublishSubject;
 
@@ -360,4 +367,32 @@ public void call(Long t) {
             }});
         assertEquals(Arrays.asList(1L,1L,2L,3L), list);
     }
+    @Test
+    public void testBackpressure() {
+        final List<Long> requests = new ArrayList<Long>();
+        
+        Observable<Integer> source = Observable
+                .just(1)
+                .concatWith(Observable.<Integer>error(new TestException()))
+                .doOnRequest(new Action1<Long>() {
+                    @Override
+                    public void call(Long t) {
+                        requests.add(t);
+                    }
+                });
+        
+        TestSubscriber<Integer> ts = TestSubscriber.create(3);
+        source
+        .retry(new Func2<Integer, Throwable, Boolean>() {
+            @Override
+            public Boolean call(Integer t1, Throwable t2) {
+                return t1 < 3;
+            }
+        }).subscribe(ts);
+        
+        assertEquals(Arrays.asList(3L, 2L, 1L), requests);
+        ts.assertValues(1, 1, 1);
+        ts.assertNotCompleted();
+        ts.assertNoErrors();
+    }
 }