2.x: A.flatMapB to eagerly check for cancellations before subscribing (#4992)

akarnokd · web-flow · commit 2e6f94c35c77 · 2017-01-16T11:13:06.000+01:00
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java
@@ -152,24 +152,25 @@ public void onNext(T t) {
                 }
             } else {
                 InnerSubscriber<T, U> inner = new InnerSubscriber<T, U>(this, uniqueId++);
-                addInner(inner);
-                p.subscribe(inner);
+                if (addInner(inner)) {
+                    p.subscribe(inner);
+                }
             }
         }
 
-        void addInner(InnerSubscriber<T, U> inner) {
+        boolean addInner(InnerSubscriber<T, U> inner) {
             for (;;) {
                 InnerSubscriber<?, ?>[] a = subscribers.get();
                 if (a == CANCELLED) {
                     inner.dispose();
-                    return;
+                    return false;
                 }
                 int n = a.length;
                 InnerSubscriber<?, ?>[] b = new InnerSubscriber[n + 1];
                 System.arraycopy(a, 0, b, 0, n);
                 b[n] = inner;
                 if (subscribers.compareAndSet(a, b)) {
-                    return;
+                    return true;
                 }
             }
         }
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletable.java
@@ -116,9 +116,9 @@ public void onNext(T value) {
 
             InnerConsumer inner = new InnerConsumer();
 
-            set.add(inner);
-
-            cs.subscribe(inner);
+            if (set.add(inner)) {
+                cs.subscribe(inner);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletableCompletable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletableCompletable.java
@@ -124,9 +124,9 @@ public void onNext(T value) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            cs.subscribe(inner);
+            if (set.add(inner)) {
+                cs.subscribe(inner);
+            }
         }
 
         @Override
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,9 +128,9 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            ms.subscribe(inner);
+            if (set.add(inner)) {
+                ms.subscribe(inner);
+            }
         }
 
         @Override
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,9 +128,9 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            ms.subscribe(inner);
+            if (set.add(inner)) {
+                ms.subscribe(inner);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeFlatMapCompletable.java
@@ -87,7 +87,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            cs.subscribe(this);
+            if (!isDisposed()) {
+                cs.subscribe(this);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeFlatMapSingle.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeFlatMapSingle.java
@@ -91,7 +91,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            ss.subscribe(new FlatMapSingleObserver<R>(this, actual));
+            if (!isDisposed()) {
+                ss.subscribe(new FlatMapSingleObserver<R>(this, actual));
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeFlatten.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeFlatten.java
@@ -93,7 +93,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            source.subscribe(new InnerObserver());
+            if (!isDisposed()) {
+                source.subscribe(new InnerObserver());
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMap.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMap.java
@@ -158,26 +158,27 @@ void subscribeInner(ObservableSource<? extends U> p) {
                     }
                 } else {
                     InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
-                    addInner(inner);
-                    p.subscribe(inner);
+                    if (addInner(inner)) {
+                        p.subscribe(inner);
+                    }
                     break;
                 }
             }
         }
 
-        void addInner(InnerObserver<T, U> inner) {
+        boolean addInner(InnerObserver<T, U> inner) {
             for (;;) {
                 InnerObserver<?, ?>[] a = observers.get();
                 if (a == CANCELLED) {
                     inner.dispose();
-                    return;
+                    return false;
                 }
                 int n = a.length;
                 InnerObserver<?, ?>[] b = new InnerObserver[n + 1];
                 System.arraycopy(a, 0, b, 0, n);
                 b[n] = inner;
                 if (observers.compareAndSet(a, b)) {
-                    return;
+                    return true;
                 }
             }
         }
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletable.java
@@ -98,9 +98,9 @@ public void onNext(T value) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            cs.subscribe(inner);
+            if (set.add(inner)) {
+                cs.subscribe(inner);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletableCompletable.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapCompletableCompletable.java
@@ -104,9 +104,9 @@ public void onNext(T value) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            cs.subscribe(inner);
+            if (set.add(inner)) {
+                cs.subscribe(inner);
+            }
         }
 
         @Override
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,9 +109,9 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            ms.subscribe(inner);
+            if (set.add(inner)) {
+                ms.subscribe(inner);
+            }
         }
 
         @Override
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,9 +109,9 @@ public void onNext(T t) {
 
             InnerObserver inner = new InnerObserver();
 
-            set.add(inner);
-
-            ms.subscribe(inner);
+            if (set.add(inner)) {
+                ms.subscribe(inner);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleFlatMap.java b/src/main/java/io/reactivex/internal/operators/single/SingleFlatMap.java
@@ -80,7 +80,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            o.subscribe(new FlatMapSingleObserver<R>(this, actual));
+            if (!isDisposed()) {
+                o.subscribe(new FlatMapSingleObserver<R>(this, actual));
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/single/SingleFlatMapCompletable.java
@@ -87,7 +87,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            cs.subscribe(this);
+            if (!isDisposed()) {
+                cs.subscribe(this);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleFlatMapMaybe.java b/src/main/java/io/reactivex/internal/operators/single/SingleFlatMapMaybe.java
@@ -85,7 +85,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            ms.subscribe(new FlatMapMaybeObserver<R>(this, actual));
+            if (!isDisposed()) {
+                ms.subscribe(new FlatMapMaybeObserver<R>(this, actual));
+            }
         }
 
         @Override
diff --git a/src/test/java/io/reactivex/XFlatMapTest.java b/src/test/java/io/reactivex/XFlatMapTest.java

Original file line number	Diff line number	Diff line change
`@@ -152,24 +152,25 @@ public void onNext(T t) {`
`152`	`152`	`}`
`153`	`153`	`} else {`
`154`	`154`	`InnerSubscriber<T, U> inner = new InnerSubscriber<T, U>(this, uniqueId++);`
`155`		`- addInner(inner);`
`156`		`- p.subscribe(inner);`
	`155`	`+ if (addInner(inner)) {`
	`156`	`+ p.subscribe(inner);`
	`157`	`+ }`
`157`	`158`	`}`
`158`	`159`	`}`
`159`	`160`
`160`		`- void addInner(InnerSubscriber<T, U> inner) {`
	`161`	`+ boolean addInner(InnerSubscriber<T, U> inner) {`
`161`	`162`	`for (;;) {`
`162`	`163`	`InnerSubscriber<?, ?>[] a = subscribers.get();`
`163`	`164`	`if (a == CANCELLED) {`
`164`	`165`	`inner.dispose();`
`165`		`- return;`
	`166`	`+ return false;`
`166`	`167`	`}`
`167`	`168`	`int n = a.length;`
`168`	`169`	`InnerSubscriber<?, ?>[] b = new InnerSubscriber[n + 1];`
`169`	`170`	`System.arraycopy(a, 0, b, 0, n);`
`170`	`171`	`b[n] = inner;`
`171`	`172`	`if (subscribers.compareAndSet(a, b)) {`
`172`		`- return;`
	`173`	`+ return true;`
`173`	`174`	`}`
`174`	`175`	`}`
`175`	`176`	`}`
Original file line number	Diff line number	Diff line change
`@@ -87,7 +87,9 @@ public void onSuccess(T value) {`
`87`	`87`	`return;`
`88`	`88`	`}`
`89`	`89`
`90`		`- cs.subscribe(this);`
	`90`	`+ if (!isDisposed()) {`
	`91`	`+ cs.subscribe(this);`
	`92`	`+ }`
`91`	`93`	`}`
`92`	`94`
`93`	`95`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,9 @@ public void onSuccess(T value) {`
`91`	`91`	`return;`
`92`	`92`	`}`
`93`	`93`
`94`		`- ss.subscribe(new FlatMapSingleObserver<R>(this, actual));`
	`94`	`+ if (!isDisposed()) {`
	`95`	`+ ss.subscribe(new FlatMapSingleObserver<R>(this, actual));`
	`96`	`+ }`
`95`	`97`	`}`
`96`	`98`
`97`	`99`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,9 @@ public void onSuccess(T value) {`
`93`	`93`	`return;`
`94`	`94`	`}`
`95`	`95`
`96`		`- source.subscribe(new InnerObserver());`
	`96`	`+ if (!isDisposed()) {`
	`97`	`+ source.subscribe(new InnerObserver());`
	`98`	`+ }`
`97`	`99`	`}`
`98`	`100`
`99`	`101`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -158,26 +158,27 @@ void subscribeInner(ObservableSource<? extends U> p) {`
`158`	`158`	`}`
`159`	`159`	`} else {`
`160`	`160`	`InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);`
`161`		`- addInner(inner);`
`162`		`- p.subscribe(inner);`
	`161`	`+ if (addInner(inner)) {`
	`162`	`+ p.subscribe(inner);`
	`163`	`+ }`
`163`	`164`	`break;`
`164`	`165`	`}`
`165`	`166`	`}`
`166`	`167`	`}`
`167`	`168`
`168`		`- void addInner(InnerObserver<T, U> inner) {`
	`169`	`+ boolean addInner(InnerObserver<T, U> inner) {`
`169`	`170`	`for (;;) {`
`170`	`171`	`InnerObserver<?, ?>[] a = observers.get();`
`171`	`172`	`if (a == CANCELLED) {`
`172`	`173`	`inner.dispose();`
`173`		`- return;`
	`174`	`+ return false;`
`174`	`175`	`}`
`175`	`176`	`int n = a.length;`
`176`	`177`	`InnerObserver<?, ?>[] b = new InnerObserver[n + 1];`
`177`	`178`	`System.arraycopy(a, 0, b, 0, n);`
`178`	`179`	`b[n] = inner;`
`179`	`180`	`if (observers.compareAndSet(a, b)) {`
`180`		`- return;`
	`181`	`+ return true;`
`181`	`182`	`}`
`182`	`183`	`}`
`183`	`184`	`}`