2.x: Fix concatMapSingle & concatMapMaybe dispose-cleanup crash (#5928)

ReactiveX · Mar 23, 2018 · 0fe6e55 · 0fe6e55
1 parent 63572c7
commit 0fe6e55
Show file tree

Hide file tree

Showing 8 changed files with 108 additions and 4 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapMaybe.java b/src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapMaybe.java
@@ -170,7 +170,7 @@ public void cancel() {
             cancelled = true;
             upstream.cancel();
             inner.dispose();
-            if (getAndIncrement() != 0) {
+            if (getAndIncrement() == 0) {
                 queue.clear();
                 item = null;
             }

diff --git a/src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingle.java b/src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingle.java
@@ -170,7 +170,7 @@ public void cancel() {
             cancelled = true;
             upstream.cancel();
             inner.dispose();
-            if (getAndIncrement() != 0) {
+            if (getAndIncrement() == 0) {
                 queue.clear();
                 item = null;
             }

diff --git a/src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybe.java b/src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybe.java
@@ -148,7 +148,7 @@ public void dispose() {
             cancelled = true;
             upstream.dispose();
             inner.dispose();
-            if (getAndIncrement() != 0) {
+            if (getAndIncrement() == 0) {
                 queue.clear();
                 item = null;
             }

diff --git a/src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingle.java b/src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingle.java
@@ -148,7 +148,7 @@ public void dispose() {
             cancelled = true;
             upstream.dispose();
             inner.dispose();
-            if (getAndIncrement() != 0) {
+            if (getAndIncrement() == 0) {
                 queue.clear();
                 item = null;
             }

diff --git a/src/test/java/io/reactivex/internal/operators/mixed/FlowableConcatMapMaybeTest.java b/src/test/java/io/reactivex/internal/operators/mixed/FlowableConcatMapMaybeTest.java
@@ -27,7 +27,9 @@
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.operators.mixed.FlowableConcatMapMaybe.ConcatMapMaybeSubscriber;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
+import io.reactivex.internal.util.ErrorMode;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.PublishProcessor;
 import io.reactivex.schedulers.Schedulers;
@@ -368,4 +370,28 @@ public MaybeSource<? extends Object> apply(Integer v)
 
         assertFalse(pp.hasSubscribers());
     }
+
+    @Test(timeout = 10000)
+    public void cancelNoConcurrentClean() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ConcatMapMaybeSubscriber<Integer, Integer> operator =
+                new ConcatMapMaybeSubscriber<Integer, Integer>(
+                        ts, Functions.justFunction(Maybe.<Integer>never()), 16, ErrorMode.IMMEDIATE);
+
+        operator.onSubscribe(new BooleanSubscription());
+
+        operator.queue.offer(1);
+
+        operator.getAndIncrement();
+
+        ts.cancel();
+
+        assertFalse(operator.queue.isEmpty());
+
+        operator.addAndGet(-2);
+
+        operator.cancel();
+
+        assertTrue(operator.queue.isEmpty());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingleTest.java b/src/test/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingleTest.java
@@ -26,7 +26,9 @@
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.operators.mixed.FlowableConcatMapSingle.ConcatMapSingleSubscriber;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
+import io.reactivex.internal.util.ErrorMode;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.PublishProcessor;
 import io.reactivex.subjects.SingleSubject;
@@ -283,4 +285,28 @@ public SingleSource<? extends Object> apply(Integer v)
 
         assertFalse(pp.hasSubscribers());
     }
+
+    @Test(timeout = 10000)
+    public void cancelNoConcurrentClean() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ConcatMapSingleSubscriber<Integer, Integer> operator =
+                new ConcatMapSingleSubscriber<Integer, Integer>(
+                        ts, Functions.justFunction(Single.<Integer>never()), 16, ErrorMode.IMMEDIATE);
+
+        operator.onSubscribe(new BooleanSubscription());
+
+        operator.queue.offer(1);
+
+        operator.getAndIncrement();
+
+        ts.cancel();
+
+        assertFalse(operator.queue.isEmpty());
+
+        operator.addAndGet(-2);
+
+        operator.cancel();
+
+        assertTrue(operator.queue.isEmpty());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybeTest.java b/src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybeTest.java
@@ -26,6 +26,8 @@
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.operators.mixed.ObservableConcatMapMaybe.ConcatMapMaybeMainObserver;
+import io.reactivex.internal.util.ErrorMode;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
@@ -373,4 +375,28 @@ public void scalarEmptySource() {
 
         assertFalse(ms.hasObservers());
     }
+
+    @Test(timeout = 10000)
+    public void cancelNoConcurrentClean() {
+        TestObserver<Integer> to = new TestObserver<Integer>();
+        ConcatMapMaybeMainObserver<Integer, Integer> operator =
+                new ConcatMapMaybeMainObserver<Integer, Integer>(
+                        to, Functions.justFunction(Maybe.<Integer>never()), 16, ErrorMode.IMMEDIATE);
+
+        operator.onSubscribe(Disposables.empty());
+
+        operator.queue.offer(1);
+
+        operator.getAndIncrement();
+
+        to.dispose();
+
+        assertFalse(operator.queue.isEmpty());
+
+        operator.addAndGet(-2);
+
+        operator.dispose();
+
+        assertTrue(operator.queue.isEmpty());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingleTest.java b/src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingleTest.java
@@ -25,6 +25,8 @@
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.operators.mixed.ObservableConcatMapSingle.ConcatMapSingleMainObserver;
+import io.reactivex.internal.util.ErrorMode;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.*;
@@ -310,4 +312,28 @@ public void scalarEmptySource() {
 
         assertFalse(ss.hasObservers());
     }
+
+    @Test(timeout = 10000)
+    public void cancelNoConcurrentClean() {
+        TestObserver<Integer> to = new TestObserver<Integer>();
+        ConcatMapSingleMainObserver<Integer, Integer> operator =
+                new ConcatMapSingleMainObserver<Integer, Integer>(
+                        to, Functions.justFunction(Single.<Integer>never()), 16, ErrorMode.IMMEDIATE);
+
+        operator.onSubscribe(Disposables.empty());
+
+        operator.queue.offer(1);
+
+        operator.getAndIncrement();
+
+        to.cancel();
+
+        assertFalse(operator.queue.isEmpty());
+
+        operator.addAndGet(-2);
+
+        operator.dispose();
+
+        assertTrue(operator.queue.isEmpty());
+    }
 }