2.x: more eager cancellation in flatMapX (#5422)

* 2.x: more eager cancellation in flatMapX * Add more eager check to Observable.flatMapX
ReactiveX · Jun 17, 2017 · 73a85c1 · 73a85c1
1 parent b7086ef
commit 73a85c1
Show file tree

Hide file tree

Showing 8 changed files with 20 additions and 8 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletable.java
@@ -72,6 +72,8 @@ static final class FlatMapCompletableMainSubscriber<T> extends BasicIntQueueSubs
 
         Subscription s;
 
+        volatile boolean cancelled;
+
         FlatMapCompletableMainSubscriber(Subscriber<? super T> observer,
                 Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors,
                 int maxConcurrency) {
@@ -117,7 +119,7 @@ public void onNext(T value) {
 
             InnerConsumer inner = new InnerConsumer();
 
-            if (set.add(inner)) {
+            if (!cancelled && set.add(inner)) {
                 cs.subscribe(inner);
             }
         }
@@ -164,6 +166,7 @@ public void onComplete() {
 
         @Override
         public void cancel() {
+            cancelled = true;
             s.cancel();
             set.dispose();
         }

diff --git a/.../java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletableCompletable.java b/.../java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletableCompletable.java
@@ -79,6 +79,8 @@ static final class FlatMapCompletableMainSubscriber<T> extends AtomicInteger
 
         Subscription s;
 
+        volatile boolean disposed;
+
         FlatMapCompletableMainSubscriber(CompletableObserver observer,
                 Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors,
                 int maxConcurrency) {
@@ -124,7 +126,7 @@ public void onNext(T value) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!disposed && set.add(inner)) {
                 cs.subscribe(inner);
             }
         }
@@ -171,6 +173,7 @@ public void onComplete() {
 
         @Override
         public void dispose() {
+            disposed = true;
             s.cancel();
             set.dispose();
         }

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapMaybe.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapMaybe.java
@@ -128,7 +128,7 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!cancelled && set.add(inner)) {
                 ms.subscribe(inner);
             }
         }

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapSingle.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapSingle.java
@@ -128,7 +128,7 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!cancelled && set.add(inner)) {
                 ms.subscribe(inner);
             }
         }

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletable.java
@@ -64,6 +64,8 @@ static final class FlatMapCompletableMainObserver<T> extends BasicIntQueueDispos
 
         Disposable d;
 
+        volatile boolean disposed;
+
         FlatMapCompletableMainObserver(Observer<? super T> observer, Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {
             this.actual = observer;
             this.mapper = mapper;
@@ -99,7 +101,7 @@ public void onNext(T value) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!disposed && set.add(inner)) {
                 cs.subscribe(inner);
             }
         }
@@ -138,6 +140,7 @@ public void onComplete() {
 
         @Override
         public void dispose() {
+            disposed = true;
             d.dispose();
             set.dispose();
         }

diff --git a/...a/io/reactivex/internal/operators/observable/ObservableFlatMapCompletableCompletable.java b/...a/io/reactivex/internal/operators/observable/ObservableFlatMapCompletableCompletable.java
@@ -69,6 +69,8 @@ static final class FlatMapCompletableMainObserver<T> extends AtomicInteger imple
 
         Disposable d;
 
+        volatile boolean disposed;
+
         FlatMapCompletableMainObserver(CompletableObserver observer, Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {
             this.actual = observer;
             this.mapper = mapper;
@@ -104,7 +106,7 @@ public void onNext(T value) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!disposed && set.add(inner)) {
                 cs.subscribe(inner);
             }
         }
@@ -143,6 +145,7 @@ public void onComplete() {
 
         @Override
         public void dispose() {
+            disposed = true;
             d.dispose();
             set.dispose();
         }

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapMaybe.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapMaybe.java
@@ -109,7 +109,7 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!cancelled && set.add(inner)) {
                 ms.subscribe(inner);
             }
         }

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapSingle.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapSingle.java
@@ -109,7 +109,7 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            if (set.add(inner)) {
+            if (!cancelled && set.add(inner)) {
                 ms.subscribe(inner);
             }
         }