2.x: TestObserver shouldn't clear the upstream disposable on terminated

Author: akarnokd

src/main/java/io/reactivex/observers/TestObserver.java

@@ -173,8 +173,6 @@ public void onError(Throwable t) {
             }
 
             actual.onError(t);
-
-            subscription.lazySet(DisposableHelper.DISPOSED);
         } finally {
             done.countDown();
         }
@@ -194,8 +192,6 @@ public void onComplete() {
             completions++;
 
             actual.onComplete();
-
-            subscription.lazySet(DisposableHelper.DISPOSED);
         } finally {
             done.countDown();
         }

src/test/java/io/reactivex/internal/operators/completable/CompletableDoOnTest.java

@@ -14,12 +14,14 @@
 package io.reactivex.internal.operators.completable;
 
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.Test;
+import static org.junit.Assert.*;
+import org.junit.*;
 
 import io.reactivex.*;
 import io.reactivex.exceptions.*;
-import io.reactivex.functions.Consumer;
+import io.reactivex.functions.*;
 import io.reactivex.observers.TestObserver;
 
 public class CompletableDoOnTest {
@@ -52,4 +54,24 @@ public void accept(Throwable e) throws Exception {
         TestHelper.assertError(errors, 0, TestException.class, "Outer");
         TestHelper.assertError(errors, 1, TestException.class, "Inner");
     }
+
+    @Test
+    public void doOnDisposeCalled() {
+        final AtomicBoolean atomicBoolean = new AtomicBoolean();
+
+        assertFalse(atomicBoolean.get());
+
+        Completable.complete()
+            .doOnDispose(new Action() {
+                @Override
+                public void run() throws Exception {
+                    atomicBoolean.set(true);
+                }
+            })
+            .test()
+            .assertResult()
+            .dispose();
+
+        assertTrue(atomicBoolean.get());
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableTakeUntilTest.java

@@ -205,7 +205,8 @@ public void testUntilFires() {
 
         assertFalse("Source still has observers", source.hasObservers());
         assertFalse("Until still has observers", until.hasObservers());
-        assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+//        assertTrue("Not cancelled!", ts.isCancelled());
     }
     @Test
     public void testMainCompletes() {
@@ -228,7 +229,8 @@ public void testMainCompletes() {
 
         assertFalse("Source still has observers", source.hasObservers());
         assertFalse("Until still has observers", until.hasObservers());
-        assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+//        assertTrue("Not cancelled!", ts.isCancelled());
     }
     @Test
     public void testDownstreamUnsubscribes() {
@@ -250,7 +252,8 @@ public void testDownstreamUnsubscribes() {
 
         assertFalse("Source still has observers", source.hasObservers());
         assertFalse("Until still has observers", until.hasObservers());
-        assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+//        assertTrue("Not cancelled!", ts.isCancelled());
     }
 
     @Test

src/test/java/io/reactivex/internal/operators/observable/ObservableTakeWhileTest.java

@@ -228,7 +228,8 @@ public boolean test(Integer t1) {
         ts.assertNoErrors();
         ts.assertValue(1);
 
-        Assert.assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+//        Assert.assertTrue("Not cancelled!", ts.isCancelled());
     }
 
     @Test

src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithObservableTest.java

@@ -296,7 +296,9 @@ public Observable<String> call() {
         TestObserver<Observable<Integer>> ts = new TestObserver<Observable<Integer>>();
         source.window(boundary).subscribe(ts);
 
-        assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+        // assertTrue("Not cancelled!", ts.isCancelled());
+        ts.assertComplete();
     }
 
     @Test

src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithStartEndObservableTest.java

@@ -226,7 +226,8 @@ public Observable<Integer> apply(Integer t) {
         ts.assertNoErrors();
         ts.assertValueCount(1);
 
-        assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+//        assertTrue("Not cancelled!", ts.isCancelled());
         assertFalse(open.hasObservers());
         assertFalse(close.hasObservers());
     }

src/test/java/io/reactivex/internal/operators/observable/ObservableWithLatestFromTest.java

@@ -255,7 +255,8 @@ public void testNoDownstreamUnsubscribe() {
 
         source.onComplete();
 
-        assertTrue("Not cancelled!", ts.isCancelled());
+        // 2.0.2 - not anymore
+//        assertTrue("Not cancelled!", ts.isCancelled());
     }
 
 

src/test/java/io/reactivex/observers/TestObserverTest.java

@@ -1175,14 +1175,16 @@ public void asyncQueueThrows() {
 
     @Test
     public void completedMeansDisposed() {
-        assertTrue(Observable.just(1)
+        // 2.0.2 - a terminated TestObserver no longer reports isDisposed
+        assertFalse(Observable.just(1)
                 .test()
                 .assertResult(1).isDisposed());
     }
 
     @Test
     public void errorMeansDisposed() {
-        assertTrue(Observable.error(new TestException())
+        // 2.0.2 - a terminated TestObserver no longer reports isDisposed
+        assertFalse(Observable.error(new TestException())
                 .test()
                 .assertFailure(TestException.class).isDisposed());
     }