2.x: Single.subscribe(BiConsumer) consistent isDisposed (#5277)

Author: akarnokd

src/main/java/io/reactivex/internal/observers/BiConsumerSingleObserver.java

@@ -37,6 +37,7 @@ public BiConsumerSingleObserver(BiConsumer<? super T, ? super Throwable> onCallb
     @Override
     public void onError(Throwable e) {
         try {
+            lazySet(DisposableHelper.DISPOSED);
             onCallback.accept(null, e);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
@@ -52,6 +53,7 @@ public void onSubscribe(Disposable d) {
     @Override
     public void onSuccess(T value) {
         try {
+            lazySet(DisposableHelper.DISPOSED);
             onCallback.accept(value, null);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);

src/test/java/io/reactivex/single/SingleSubscribeTest.java

@@ -15,6 +15,7 @@
 
 import static org.junit.Assert.*;
 
+import java.io.IOException;
 import java.util.List;
 
 import org.junit.Test;
@@ -227,4 +228,40 @@ public void successIsDisposed() {
     public void errorIsDisposed() {
         assertTrue(Single.error(new TestException()).subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()).isDisposed());
     }
+
+    @Test
+    public void biConsumerIsDisposedOnSuccess() {
+        final Object[] result = { null, null };
+        
+        Disposable d = Single.just(1)
+        .subscribe(new BiConsumer<Integer, Throwable>() {
+            @Override
+            public void accept(Integer t1, Throwable t2) throws Exception {
+                result[0] = t1;
+                result[1] = t2;
+            }
+        });
+        
+        assertTrue("Not disposed?!", d.isDisposed());
+        assertEquals(1, result[0]);
+        assertNull(result[1]);
+    }
+
+    @Test
+    public void biConsumerIsDisposedOnError() {
+        final Object[] result = { null, null };
+        
+        Disposable d = Single.<Integer>error(new IOException())
+        .subscribe(new BiConsumer<Integer, Throwable>() {
+            @Override
+            public void accept(Integer t1, Throwable t2) throws Exception {
+                result[0] = t1;
+                result[1] = t2;
+            }
+        });
+        
+        assertTrue("Not disposed?!", d.isDisposed());
+        assertNull(result[0]);
+        assertTrue("" + result[1], result[1] instanceof IOException);
+    }
 }