1.x: make Completable.subscribe() report isUnsubscribed consistently

ReactiveX · Feb 15, 2016 · 58826fa · 58826fa
1 parent ee9956a
commit 58826fa
Show file tree

Hide file tree

Showing 2 changed files with 145 additions and 1 deletion.
diff --git a/src/main/java/rx/Completable.java b/src/main/java/rx/Completable.java
@@ -1828,11 +1828,12 @@ public final Subscription subscribe() {
         subscribe(new CompletableSubscriber() {
             @Override
             public void onCompleted() {
-                // nothing to do
+                mad.unsubscribe();
             }
 
             @Override
             public void onError(Throwable e) {
+                mad.unsubscribe();
                 ERROR_HANDLER.handleError(e);
             }
 
@@ -1864,10 +1865,12 @@ public void onCompleted() {
                 } catch (Throwable e) {
                     ERROR_HANDLER.handleError(e);
                 }
+                mad.unsubscribe();
             }
 
             @Override
             public void onError(Throwable e) {
+                mad.unsubscribe();
                 ERROR_HANDLER.handleError(e);
             }
 
@@ -1900,7 +1903,9 @@ public void onCompleted() {
                     onComplete.call();
                 } catch (Throwable e) {
                     onError(e);
+                    return;
                 }
+                mad.unsubscribe();
             }
 
             @Override
@@ -1911,6 +1916,7 @@ public void onError(Throwable e) {
                     e = new CompositeException(Arrays.asList(e, ex));
                     ERROR_HANDLER.handleError(e);
                 }
+                mad.unsubscribe();
             }
 
             @Override

diff --git a/src/test/java/rx/CompletableTest.java b/src/test/java/rx/CompletableTest.java
@@ -3604,4 +3604,142 @@ public Completable call(Integer t) {
         assertTrue(listEx.get(1).toString(), listEx.get(1) instanceof TestException);
     }
 
+    @Test
+    public void subscribeReportsUnsubscribed() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        Subscription completableSubscription = completable.subscribe();
+
+        stringSubject.onCompleted();
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+    }
+
+    @Test
+    public void subscribeReportsUnsubscribedOnError() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        Subscription completableSubscription = completable.subscribe();
+
+        stringSubject.onError(new TestException());
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+    }
+
+    @Test
+    public void subscribeActionReportsUnsubscribed() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        Subscription completableSubscription = completable.subscribe(Actions.empty());
+
+        stringSubject.onCompleted();
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+    }
+
+    @Test
+    public void subscribeActionReportsUnsubscribedAfter() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        final AtomicReference<Subscription> subscriptionRef = new AtomicReference<Subscription>();
+        Subscription completableSubscription = completable.subscribe(new Action0() {
+            @Override
+            public void call() {
+                if (subscriptionRef.get().isUnsubscribed()) {
+                    subscriptionRef.set(null);
+                }
+            }
+        });
+        subscriptionRef.set(completableSubscription);
+
+        stringSubject.onCompleted();
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+        assertNotNull("Unsubscribed before the call to onCompleted", subscriptionRef.get());
+    }
+
+    @Test
+    public void subscribeActionReportsUnsubscribedOnError() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        Subscription completableSubscription = completable.subscribe(Actions.empty());
+
+        stringSubject.onError(new TestException());
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+    }
+
+    @Test
+    public void subscribeAction2ReportsUnsubscribed() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        Subscription completableSubscription = completable.subscribe(Actions.empty(), Actions.empty());
+
+        stringSubject.onCompleted();
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+    }
+
+    @Test
+    public void subscribeAction2ReportsUnsubscribedOnError() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        Subscription completableSubscription = completable.subscribe(Actions.empty(), Actions.empty());
+
+        stringSubject.onError(new TestException());
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+    }
+
+    @Test
+    public void subscribeAction2ReportsUnsubscribedAfter() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        final AtomicReference<Subscription> subscriptionRef = new AtomicReference<Subscription>();
+        Subscription completableSubscription = completable.subscribe(Actions.empty(), new Action0() {
+            @Override
+            public void call() {
+                if (subscriptionRef.get().isUnsubscribed()) {
+                    subscriptionRef.set(null);
+                }
+            }
+        });
+        subscriptionRef.set(completableSubscription);
+
+        stringSubject.onCompleted();
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+        assertNotNull("Unsubscribed before the call to onCompleted", subscriptionRef.get());
+    }
+
+    @Test
+    public void subscribeAction2ReportsUnsubscribedOnErrorAfter() {
+        PublishSubject<String> stringSubject = PublishSubject.create();
+        Completable completable = stringSubject.toCompletable();
+
+        final AtomicReference<Subscription> subscriptionRef = new AtomicReference<Subscription>();
+        Subscription completableSubscription = completable.subscribe(new Action1<Throwable>() {
+            @Override
+            public void call(Throwable e) {
+                if (subscriptionRef.get().isUnsubscribed()) {
+                    subscriptionRef.set(null);
+                }
+            }
+        }, Actions.empty());
+        subscriptionRef.set(completableSubscription);
+
+        stringSubject.onError(new TestException());
+
+        assertTrue("Not unsubscribed?", completableSubscription.isUnsubscribed());
+        assertNotNull("Unsubscribed before the call to onError", subscriptionRef.get());
+    }
+
 }