3.x: Fix map() conditional chain causing NPE (#7040)

Author: akarnokd

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableMap.java

@@ -115,7 +115,12 @@ public void onNext(T t) {
         @Override
         public boolean tryOnNext(T t) {
             if (done) {
-                return false;
+                return true;
+            }
+
+            if (sourceMode != NONE) {
+                downstream.tryOnNext(null);
+                return true;
             }
 
             U v;

src/test/java/io/reactivex/rxjava3/internal/jdk8/FlowableMapOptionalTest.java

@@ -24,9 +24,10 @@
 import io.reactivex.rxjava3.exceptions.TestException;
 import io.reactivex.rxjava3.functions.Function;
 import io.reactivex.rxjava3.internal.fuseable.QueueFuseable;
+import io.reactivex.rxjava3.internal.schedulers.ImmediateThinScheduler;
 import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription;
 import io.reactivex.rxjava3.processors.*;
-import io.reactivex.rxjava3.testsupport.TestHelper;
+import io.reactivex.rxjava3.testsupport.*;
 
 public class FlowableMapOptionalTest extends RxJavaTest {
 
@@ -467,4 +468,20 @@ public void boundaryFusedMixedConditional() {
         .assertFusionMode(QueueFuseable.NONE)
         .assertResult(2, 4, 6, 8, 10);
     }
+
+    @Test
+    public void conditionalFusionNoNPE() {
+        TestSubscriberEx<Object> ts = new TestSubscriberEx<>()
+        .setInitialFusionMode(QueueFuseable.ANY);
+
+        Flowable.empty()
+        .observeOn(ImmediateThinScheduler.INSTANCE)
+        .filter(v -> true)
+        .mapOptional(Optional::of)
+        .filter(v -> true)
+        .subscribe(ts)
+        ;
+
+        ts.assertResult();
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableMapTest.java

@@ -28,6 +28,7 @@
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.functions.Functions;
 import io.reactivex.rxjava3.internal.fuseable.*;
+import io.reactivex.rxjava3.internal.schedulers.ImmediateThinScheduler;
 import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 import io.reactivex.rxjava3.processors.*;
@@ -617,4 +618,19 @@ public Object apply(Flowable<Object> f) throws Exception {
         }, false, 1, 1, 1);
     }
 
+    @Test
+    public void conditionalFusionNoNPE() {
+        TestSubscriberEx<Object> ts = new TestSubscriberEx<>()
+        .setInitialFusionMode(QueueFuseable.ANY);
+
+        Flowable.empty()
+        .observeOn(ImmediateThinScheduler.INSTANCE)
+        .filter(v -> true)
+        .map(v -> v)
+        .filter(v -> true)
+        .subscribe(ts)
+        ;
+
+        ts.assertResult();
+    }
 }