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

akarnokd · web-flow · commit 73a85c15d35d · 2017-06-17T08:23:44.000+02:00
* 2.x: more eager cancellation in flatMapX

* Add more eager check to Observable.flatMapX
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/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletableCompletable.java b/src/main/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/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletableCompletable.java b/src/main/java/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);
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -72,6 +72,8 @@ static final class FlatMapCompletableMainSubscriber<T> extends BasicIntQueueSubs`
`72`	`72`
`73`	`73`	`Subscription s;`
`74`	`74`
	`75`	`+ volatile boolean cancelled;`
	`76`	`+`
`75`	`77`	`FlatMapCompletableMainSubscriber(Subscriber<? super T> observer,`
`76`	`78`	`Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors,`
`77`	`79`	`int maxConcurrency) {`
`@@ -117,7 +119,7 @@ public void onNext(T value) {`
`117`	`119`
`118`	`120`	`InnerConsumer inner = new InnerConsumer();`
`119`	`121`
`120`		`- if (set.add(inner)) {`
	`122`	`+ if (!cancelled && set.add(inner)) {`
`121`	`123`	`cs.subscribe(inner);`
`122`	`124`	`}`
`123`	`125`	`}`
`@@ -164,6 +166,7 @@ public void onComplete() {`
`164`	`166`
`165`	`167`	`@Override`
`166`	`168`	`public void cancel() {`
	`169`	`+ cancelled = true;`
`167`	`170`	`s.cancel();`
`168`	`171`	`set.dispose();`
`169`	`172`	`}`
Original file line number	Diff line number	Diff line change
`@@ -79,6 +79,8 @@ static final class FlatMapCompletableMainSubscriber<T> extends AtomicInteger`
`79`	`79`
`80`	`80`	`Subscription s;`
`81`	`81`
	`82`	`+ volatile boolean disposed;`
	`83`	`+`
`82`	`84`	`FlatMapCompletableMainSubscriber(CompletableObserver observer,`
`83`	`85`	`Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors,`
`84`	`86`	`int maxConcurrency) {`
`@@ -124,7 +126,7 @@ public void onNext(T value) {`
`124`	`126`
`125`	`127`	`InnerObserver inner = new InnerObserver();`
`126`	`128`
`127`		`- if (set.add(inner)) {`
	`129`	`+ if (!disposed && set.add(inner)) {`
`128`	`130`	`cs.subscribe(inner);`
`129`	`131`	`}`
`130`	`132`	`}`
`@@ -171,6 +173,7 @@ public void onComplete() {`
`171`	`173`
`172`	`174`	`@Override`
`173`	`175`	`public void dispose() {`
	`176`	`+ disposed = true;`
`174`	`177`	`s.cancel();`
`175`	`178`	`set.dispose();`
`176`	`179`	`}`
Original file line number	Diff line number	Diff line change
`@@ -128,7 +128,7 @@ public void onNext(T t) {`
`128`	`128`
`129`	`129`	`InnerObserver inner = new InnerObserver();`
`130`	`130`
`131`		`- if (set.add(inner)) {`
	`131`	`+ if (!cancelled && set.add(inner)) {`
`132`	`132`	`ms.subscribe(inner);`
`133`	`133`	`}`
`134`	`134`	`}`
Original file line number	Diff line number	Diff line change
`@@ -64,6 +64,8 @@ static final class FlatMapCompletableMainObserver<T> extends BasicIntQueueDispos`
`64`	`64`
`65`	`65`	`Disposable d;`
`66`	`66`
	`67`	`+ volatile boolean disposed;`
	`68`	`+`
`67`	`69`	`FlatMapCompletableMainObserver(Observer<? super T> observer, Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {`
`68`	`70`	`this.actual = observer;`
`69`	`71`	`this.mapper = mapper;`
`@@ -99,7 +101,7 @@ public void onNext(T value) {`
`99`	`101`
`100`	`102`	`InnerObserver inner = new InnerObserver();`
`101`	`103`
`102`		`- if (set.add(inner)) {`
	`104`	`+ if (!disposed && set.add(inner)) {`
`103`	`105`	`cs.subscribe(inner);`
`104`	`106`	`}`
`105`	`107`	`}`
`@@ -138,6 +140,7 @@ public void onComplete() {`
`138`	`140`
`139`	`141`	`@Override`
`140`	`142`	`public void dispose() {`
	`143`	`+ disposed = true;`
`141`	`144`	`d.dispose();`
`142`	`145`	`set.dispose();`
`143`	`146`	`}`
Original file line number	Diff line number	Diff line change
`@@ -69,6 +69,8 @@ static final class FlatMapCompletableMainObserver<T> extends AtomicInteger imple`
`69`	`69`
`70`	`70`	`Disposable d;`
`71`	`71`
	`72`	`+ volatile boolean disposed;`
	`73`	`+`
`72`	`74`	`FlatMapCompletableMainObserver(CompletableObserver observer, Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {`
`73`	`75`	`this.actual = observer;`
`74`	`76`	`this.mapper = mapper;`
`@@ -104,7 +106,7 @@ public void onNext(T value) {`
`104`	`106`
`105`	`107`	`InnerObserver inner = new InnerObserver();`
`106`	`108`
`107`		`- if (set.add(inner)) {`
	`109`	`+ if (!disposed && set.add(inner)) {`
`108`	`110`	`cs.subscribe(inner);`
`109`	`111`	`}`
`110`	`112`	`}`
`@@ -143,6 +145,7 @@ public void onComplete() {`
`143`	`145`
`144`	`146`	`@Override`
`145`	`147`	`public void dispose() {`
	`148`	`+ disposed = true;`
`146`	`149`	`d.dispose();`
`147`	`150`	`set.dispose();`
`148`	`151`	`}`
Original file line number	Diff line number	Diff line change
`@@ -109,7 +109,7 @@ public void onNext(T t) {`
`109`	`109`
`110`	`110`	`InnerObserver inner = new InnerObserver();`
`111`	`111`
`112`		`- if (set.add(inner)) {`
	`112`	`+ if (!cancelled && set.add(inner)) {`
`113`	`113`	`ms.subscribe(inner);`
`114`	`114`	`}`
`115`	`115`	`}`