2.x: Fix concatMap{Single|Maybe} null emission on dispose race (#6060)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapMaybe.java

@@ -218,6 +218,7 @@ void drain() {
                     if (cancelled) {
                         queue.clear();
                         item = null;
+                        break;
                     }
 
                     int s = state;

src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingle.java

@@ -213,6 +213,7 @@ void drain() {
                     if (cancelled) {
                         queue.clear();
                         item = null;
+                        break;
                     }
 
                     int s = state;

src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybe.java

@@ -199,6 +199,7 @@ void drain() {
                     if (cancelled) {
                         queue.clear();
                         item = null;
+                        break;
                     }
 
                     int s = state;

src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingle.java

@@ -194,6 +194,7 @@ void drain() {
                     if (cancelled) {
                         queue.clear();
                         item = null;
+                        break;
                     }
 
                     int s = state;

src/test/java/io/reactivex/internal/operators/mixed/FlowableConcatMapMaybeTest.java

@@ -394,4 +394,35 @@ public void cancelNoConcurrentClean() {
 
         assertTrue(operator.queue.isEmpty());
     }
+
+    @Test
+    public void innerSuccessDisposeRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+
+            final MaybeSubject<Integer> ms = MaybeSubject.create();
+
+            final TestSubscriber<Integer> ts = Flowable.just(1)
+                    .hide()
+                    .concatMapMaybe(Functions.justFunction(ms))
+                    .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    ms.onSuccess(1);
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ts.dispose();
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            ts.assertNoErrors();
+        }
+    }
+
 }

src/test/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingleTest.java

@@ -309,4 +309,34 @@ public void cancelNoConcurrentClean() {
 
         assertTrue(operator.queue.isEmpty());
     }
+
+    @Test
+    public void innerSuccessDisposeRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+
+            final SingleSubject<Integer> ss = SingleSubject.create();
+
+            final TestSubscriber<Integer> ts = Flowable.just(1)
+                    .hide()
+                    .concatMapSingle(Functions.justFunction(ss))
+                    .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    ss.onSuccess(1);
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ts.dispose();
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            ts.assertNoErrors();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybeTest.java

@@ -416,4 +416,34 @@ public void checkUnboundedInnerQueue() {
 
         to.assertResult(1, 2, 3, 4);
     }
+
+    @Test
+    public void innerSuccessDisposeRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+
+            final MaybeSubject<Integer> ms = MaybeSubject.create();
+
+            final TestObserver<Integer> to = Observable.just(1)
+                    .hide()
+                    .concatMapMaybe(Functions.justFunction(ms))
+                    .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    ms.onSuccess(1);
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    to.dispose();
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            to.assertNoErrors();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingleTest.java

@@ -353,4 +353,35 @@ public void checkUnboundedInnerQueue() {
 
         to.assertResult(1, 2, 3, 4);
     }
+
+    @Test
+    public void innerSuccessDisposeRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+
+            final SingleSubject<Integer> ss = SingleSubject.create();
+
+            final TestObserver<Integer> to = Observable.just(1)
+                    .hide()
+                    .concatMapSingle(Functions.justFunction(ss))
+                    .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    ss.onSuccess(1);
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    to.dispose();
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            to.assertNoErrors();
+        }
+    }
+
 }