3.x: Fix ObservableSwitchMap NPE due to cancel race

akarnokd · Aug 29, 2023 · eeef8d6 · eeef8d6
1 parent a57ad01
commit eeef8d6
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 2 deletions.
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSwitchMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSwitchMap.java
@@ -349,9 +349,10 @@ public void onSubscribe(Disposable d) {
 
         @Override
         public void onNext(R t) {
-            if (index == parent.unique) {
+            SimpleQueue<R> q = queue;
+            if (index == parent.unique && q != null) {
                 if (t != null) {
-                    queue.offer(t);
+                    q.offer(t);
                 }
                 parent.drain();
             }

diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchTest.java
@@ -1377,4 +1377,19 @@ Flowable<Integer> createFlowable(AtomicInteger inner) {
             inner.incrementAndGet();
         });
     }
+
+    @Test
+    public void innerOnSubscribeOuterCancelRace() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1)
+        .hide()
+        .switchMap(v -> Flowable.just(1)
+                .doOnSubscribe(d -> ts.cancel())
+                .scan(1, (a, b) -> a)
+        )
+        .subscribe(ts);
+
+        ts.assertEmpty();
+    }
 }
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSwitchTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSwitchTest.java
@@ -1438,4 +1438,19 @@ Observable<Integer> createObservable(AtomicInteger inner) {
             inner.incrementAndGet();
         });
     }
+
+    @Test
+    public void innerOnSubscribeOuterCancelRace() {
+        TestObserver<Integer> to = new TestObserver<Integer>();
+
+        Observable.just(1)
+        .hide()
+        .switchMap(v -> Observable.just(1)
+                .doOnSubscribe(d -> to.dispose())
+                .scan(1, (a, b) -> a)
+        )
+        .subscribe(to);
+
+        to.assertEmpty();
+    }
 }