Unit Test and Review of #1027

Author: benjchristensen

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

@@ -37,9 +37,11 @@
 import rx.Observable.Operator;
 import rx.Scheduler.Inner;
 import rx.Subscriber;
+import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.schedulers.Schedulers;
 import rx.subscriptions.SerialSubscription;
+import rx.subscriptions.Subscriptions;
 
 public class OperatorRetry<T> implements Operator<T, Observable<T>> {
 
@@ -82,6 +84,8 @@ public void call(final Inner inner) {
                         final Action1<Inner> _self = this;
                         attempts.incrementAndGet();
 
+                        // 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>() {
 
                             @Override

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

@@ -15,9 +15,13 @@
  */
 package rx.operators;
 
-import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -28,12 +32,13 @@
 import org.mockito.InOrder;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.Observable.OnSubscribe;
 import rx.Observer;
 import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action0;
 import rx.functions.Action1;
+import rx.observers.TestSubscriber;
 import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
 
@@ -114,7 +119,7 @@ public void testInfiniteRetry() {
         inOrder.verifyNoMoreInteractions();
     }
 
-    public static class FuncWithErrors implements Observable.OnSubscribeFunc<String> {
+    public static class FuncWithErrors implements Observable.OnSubscribe<String> {
 
         private final int numFailures;
         private final AtomicInteger count = new AtomicInteger(0);
@@ -124,15 +129,14 @@ public static class FuncWithErrors implements Observable.OnSubscribeFunc<String>
         }
 
         @Override
-        public Subscription onSubscribe(Observer<? super String> o) {
+        public void call(Subscriber<? 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();
         }
     }
 
@@ -153,13 +157,13 @@ public void call(Integer n) {
     }
 
     @Test
-    public void testRetryAllowsSubscriptionAfterAllSubscriptionsUnsubsribed() throws InterruptedException {
+    public void testRetryAllowsSubscriptionAfterAllSubscriptionsUnsubscribed() throws InterruptedException {
         final AtomicInteger subsCount = new AtomicInteger(0);
-        OnSubscribeFunc<String> onSubscribe = new OnSubscribeFunc<String>() {
+        OnSubscribe<String> onSubscribe = new OnSubscribe<String>() {
             @Override
-            public Subscription onSubscribe(Observer<? super String> observer) {
+            public void call(Subscriber<? super String> s) {
                 subsCount.incrementAndGet();
-                return new Subscription() {
+                s.add(new Subscription() {
                     boolean unsubscribed = false;
 
                     @Override
@@ -172,7 +176,7 @@ public void unsubscribe() {
                     public boolean isUnsubscribed() {
                         return unsubscribed;
                     }
-                };
+                });
             }
         };
         Observable<String> stream = Observable.create(onSubscribe);
@@ -185,40 +189,65 @@ public boolean isUnsubscribed() {
         assertEquals(1, subsCount.get());
     }
 
+    @Test
+    public void testSourceObservableCallsUnsubscribe() throws InterruptedException {
+        final AtomicInteger subsCount = new AtomicInteger(0);
+
+        final TestSubscriber<String> ts = new TestSubscriber<String>();
+
+        OnSubscribe<String> onSubscribe = new OnSubscribe<String>() {
+            @Override
+            public void call(Subscriber<? super String> s) {
+                // if isUnsubscribed is true that means we have a bug such as https://github.com/Netflix/RxJava/issues/1024
+                if (!s.isUnsubscribed()) {
+                    subsCount.incrementAndGet();
+                    s.onError(new RuntimeException("failed"));
+                    // it unsubscribes the child directly
+                    // this simulates various error/completion scenarios that could occur
+                    // or just a source that proactively triggers cleanup
+                    s.unsubscribe();
+                }
+            }
+        };
+
+        Observable.create(onSubscribe).retry(3).subscribe(ts);
+        assertEquals(4, subsCount.get()); // 1 + 3 retries
+    }
+
     class SlowObservable implements Observable.OnSubscribe<Long> {
 
-        private AtomicInteger efforts=new AtomicInteger(0);
-        private AtomicInteger active=new AtomicInteger(0),maxActive=new AtomicInteger(0);
+        private AtomicInteger efforts = new AtomicInteger(0);
+        private AtomicInteger active = new AtomicInteger(0), maxActive = new AtomicInteger(0);
         private AtomicInteger nextBeforeFailure;
 
         private final int emitDelay;
 
-        public SlowObservable(int emitDelay,int countNext) {
-            this.emitDelay=emitDelay;
-            this.nextBeforeFailure=new AtomicInteger(countNext);
+        public SlowObservable(int emitDelay, int countNext) {
+            this.emitDelay = emitDelay;
+            this.nextBeforeFailure = new AtomicInteger(countNext);
         }
 
         public void call(final Subscriber<? super Long> subscriber) {
-            final AtomicBoolean terminate=new AtomicBoolean(false);
+            final AtomicBoolean terminate = new AtomicBoolean(false);
             efforts.getAndIncrement();
             active.getAndIncrement();
-            maxActive.set(Math.max(active.get(),maxActive.get()));
+            maxActive.set(Math.max(active.get(), maxActive.get()));
             final Thread thread = new Thread() {
                 @Override
                 public void run() {
                     long nr = 0;
                     try {
                         while (!terminate.get()) {
                             Thread.sleep(emitDelay);
-                            if (nextBeforeFailure.getAndDecrement()>0) {
+                            if (nextBeforeFailure.getAndDecrement() > 0) {
                                 subscriber.onNext(nr++);
                             }
                             else {
                                 subscriber.onError(new RuntimeException("expected-failed"));
                             }
                         }
                     }
-                    catch(InterruptedException t) {
+                    catch (InterruptedException t) {
                     }
                 }
             };
@@ -236,13 +265,13 @@ public void call() {
     /** Observer for listener on seperate thread */
     class AsyncObserver<T> implements Observer<T> {
 
-        protected CountDownLatch latch=new CountDownLatch(1);
+        protected CountDownLatch latch = new CountDownLatch(1);
 
         protected Observer<T> target;
 
         /** Wrap existing Observer */
         public AsyncObserver(Observer<T> target) {
-            this.target=target;
+            this.target = target;
         }
 
         /** Wait */
@@ -274,19 +303,19 @@ public void onNext(T v) {
         }
     }
 
-    @Test
+    @Test(timeout = 1000)
     public void testUnsubscribeAfterError() {
 
         @SuppressWarnings("unchecked")
-        Observer<Long> observer=mock(Observer.class);
+        Observer<Long> observer = mock(Observer.class);
 
         // Observable that always fails after 100ms
-        SlowObservable so=new SlowObservable(100,0);
-        Observable<Long> o=Observable
-            .create(so)
-            .retry(5);
+        SlowObservable so = new SlowObservable(100, 0);
+        Observable<Long> o = Observable
+                .create(so)
+                .retry(5);
 
-        AsyncObserver<Long> async=new AsyncObserver<Long>(observer);
+        AsyncObserver<Long> async = new AsyncObserver<Long>(observer);
 
         o.subscribe(async);
 
@@ -297,24 +326,24 @@ public void testUnsubscribeAfterError() {
         inOrder.verify(observer, times(1)).onError(any(Throwable.class));
         inOrder.verify(observer, never()).onCompleted();
 
-        assertEquals("Start 6 threads, retry 5 then fail on 6",6,so.efforts.get());
-        assertEquals("Only 1 active subscription",1,so.maxActive.get());
+        assertEquals("Start 6 threads, retry 5 then fail on 6", 6, so.efforts.get());
+        assertEquals("Only 1 active subscription", 1, so.maxActive.get());
     }
 
-    @Test
+    @Test(timeout = 1000)
     public void testTimeoutWithRetry() {
 
         @SuppressWarnings("unchecked")
-        Observer<Long> observer=mock(Observer.class);
+        Observer<Long> observer = mock(Observer.class);
 
         // Observable that sends every 100ms (timeout fails instead)
-        SlowObservable so=new SlowObservable(100,10);
-        Observable<Long> o=Observable
-            .create(so)
-            .timeout(80, TimeUnit.MILLISECONDS)
-            .retry(5);
+        SlowObservable so = new SlowObservable(100, 10);
+        Observable<Long> o = Observable
+                .create(so)
+                .timeout(80, TimeUnit.MILLISECONDS)
+                .retry(5);
 
-        AsyncObserver<Long> async=new AsyncObserver<Long>(observer);
+        AsyncObserver<Long> async = new AsyncObserver<Long>(observer);
 
         o.subscribe(async);
 
@@ -325,6 +354,6 @@ public void testTimeoutWithRetry() {
         inOrder.verify(observer, times(1)).onError(any(Throwable.class));
         inOrder.verify(observer, never()).onCompleted();
 
-        assertEquals("Start 6 threads, retry 5 then fail on 6",6,so.efforts.get());
+        assertEquals("Start 6 threads, retry 5 then fail on 6", 6, so.efforts.get());
     }
 }