Merge pull request #1019 from benjchristensen/retry-1018

Fix: retry() never unsubscribes from source until operator completes

Author: benjchristensen

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

@@ -39,6 +39,7 @@
 import rx.Subscriber;
 import rx.functions.Action1;
 import rx.schedulers.Schedulers;
+import rx.subscriptions.SerialSubscription;
 
 public class OperatorRetry<T> implements Operator<T, Observable<T>> {
 
@@ -55,18 +56,21 @@ public OperatorRetry() {
     }
 
     @Override
-    public Subscriber<? super Observable<T>> call(final Subscriber<? super T> s) {
-        return new Subscriber<Observable<T>>(s) {
+    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 Subscriber<Observable<T>>(child) {
             final AtomicInteger attempts = new AtomicInteger(0);
-            
+
             @Override
             public void onCompleted() {
                 // ignore as we expect a single nested Observable<T>
             }
 
             @Override
             public void onError(Throwable e) {
-                s.onError(e);
+                child.onError(e);
             }
 
             @Override
@@ -77,11 +81,12 @@ public void onNext(final Observable<T> o) {
                     public void call(final Inner inner) {
                         final Action1<Inner> _self = this;
                         attempts.incrementAndGet();
-                        o.unsafeSubscribe(new Subscriber<T>(s) {
+
+                        Subscriber<T> subscriber = new Subscriber<T>(child) {
 
                             @Override
                             public void onCompleted() {
-                                s.onCompleted();
+                                child.onCompleted();
                             }
 
                             @Override
@@ -91,16 +96,19 @@ public void onError(Throwable e) {
                                     inner.schedule(_self);
                                 } else {
                                     // give up and pass the failure
-                                    s.onError(e);
+                                    child.onError(e);
                                 }
                             }
 
                             @Override
                             public void onNext(T v) {
-                                s.onNext(v);
+                                child.onNext(v);
                             }
 
-                        });
+                        };
+                        // register this Subscription (and unsubscribe previous if exists) 
+                        serialSubscription.set(subscriber);
+                        o.unsafeSubscribe(subscriber);
                     }
                 });
             }

rxjava-core/src/test/java/rx/operators/OperatorRetryTest.java

@@ -19,6 +19,8 @@
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
@@ -27,6 +29,7 @@
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action1;
 import rx.subjects.PublishSubject;
@@ -147,6 +150,93 @@ public void call(Integer n) {
         assertEquals(1, count.get());
     }
 
+    public static class SlowFuncAlwaysFails implements Observable.OnSubscribe<String> {
+
+        final AtomicInteger nextSeq=new AtomicInteger();
+        final AtomicInteger activeSubs=new AtomicInteger();
+        final AtomicInteger concurrentSubs=new AtomicInteger();
+
+        public void call(final Subscriber<? super String> s)
+        {
+            final int seq=nextSeq.incrementAndGet();
+
+            int cur=activeSubs.incrementAndGet();
+            // Track concurrent subscriptions
+            concurrentSubs.set(Math.max(cur,concurrentSubs.get()));
+
+            // Use async error
+            new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException e) {
+                        // ignore
+                    }
+                    s.onError(new RuntimeException("Subscriber #"+seq+" fails"));
+                }
+            }).start();
+
+            // Track unsubscribes
+            s.add(new Subscription()
+            {
+                private boolean active=true;
+
+                public void unsubscribe()
+                {
+                    if (active) {
+                        activeSubs.decrementAndGet();
+                        active=false;
+                    }
+                }
+
+                public boolean isUnsubscribed()
+                {
+                    return !active;
+                }
+            });
+        }
+    }
+
+    @Test
+    public void testUnsubscribeAfterError() {
+
+        final CountDownLatch check=new CountDownLatch(1);
+        final SlowFuncAlwaysFails sf=new SlowFuncAlwaysFails();
+
+        Observable
+            .create(sf)
+            .retry(4)
+            .subscribe(
+                new Action1<String>()
+                {
+                    @Override
+                    public void call(String v)
+                    {
+                        fail("Should never happen");
+                    }
+                },
+                new Action1<Throwable>()
+                {
+                    public void call(Throwable throwable)
+                    {
+                        check.countDown();
+                    }
+                }
+            );
+
+        try
+        {
+            check.await(1, TimeUnit.SECONDS);
+        } catch (InterruptedException e)
+        {
+            fail("interrupted");
+        }
+
+        assertEquals("5 Subscribers created", 5, sf.nextSeq.get());
+        assertEquals("1 Active Subscriber", 1, sf.concurrentSubs.get());
+    }
+
     @Test
     public void testRetryAllowsSubscriptionAfterAllSubscriptionsUnsubsribed() throws InterruptedException {
         final AtomicInteger subsCount = new AtomicInteger(0);