2.x: add missing null checks on values returned by user functions (#5379)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableCombineLatest.java

@@ -474,7 +474,7 @@ public R poll() throws Exception {
                 return null;
             }
             T[] a = (T[])queue.poll();
-            R r = combiner.apply(a);
+            R r = ObjectHelper.requireNonNull(combiner.apply(a), "The combiner returned a null value");
             ((CombineLatestInnerSubscriber<T>)e).requestOne();
             return r;
         }

src/main/java/io/reactivex/internal/operators/flowable/FlowableInternalHelper.java

@@ -20,7 +20,7 @@
 import io.reactivex.*;
 import io.reactivex.flowables.ConnectableFlowable;
 import io.reactivex.functions.*;
-import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.functions.*;
 
 /**
  * Helper utility class to support Flowable with inner classes.
@@ -77,7 +77,8 @@ static final class ItemDelayFunction<T, U> implements Function<T, Publisher<T>>
 
         @Override
         public Publisher<T> apply(final T v) throws Exception {
-            return new FlowableTakePublisher<U>(itemDelay.apply(v), 1).map(Functions.justFunction(v)).defaultIfEmpty(v);
+            Publisher<U> p = ObjectHelper.requireNonNull(itemDelay.apply(v), "The itemDelay returned a null Publisher");
+            return new FlowableTakePublisher<U>(p, 1).map(Functions.justFunction(v)).defaultIfEmpty(v);
         }
     }
 
@@ -164,7 +165,7 @@ static final class FlatMapWithCombinerOuter<T, R, U> implements Function<T, Publ
         @Override
         public Publisher<R> apply(final T t) throws Exception {
             @SuppressWarnings("unchecked")
-            Publisher<U> u = (Publisher<U>)mapper.apply(t);
+            Publisher<U> u = (Publisher<U>)ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Publisher");
             return new FlowableMapPublisher<U, R>(u, new FlatMapWithCombinerInner<U, R, T>(combiner, t));
         }
     }
@@ -184,7 +185,7 @@ static final class FlatMapIntoIterable<T, U> implements Function<T, Publisher<U>
 
         @Override
         public Publisher<U> apply(T t) throws Exception {
-            return new FlowableFromIterable<U>(mapper.apply(t));
+            return new FlowableFromIterable<U>(ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Iterable"));
         }
     }
 
@@ -317,7 +318,8 @@ static final class ReplayFunction<T, R> implements Function<Flowable<T>, Publish
 
         @Override
         public Publisher<R> apply(Flowable<T> t) throws Exception {
-            return Flowable.fromPublisher(selector.apply(t)).observeOn(scheduler);
+            Publisher<R> p = ObjectHelper.requireNonNull(selector.apply(t), "The selector returned a null Publisher");
+            return Flowable.fromPublisher(p).observeOn(scheduler);
         }
     }
 }

src/main/java/io/reactivex/internal/operators/flowable/FlowableMapNotification.java

@@ -18,7 +18,7 @@
 import org.reactivestreams.Subscriber;
 
 import io.reactivex.Flowable;
-import io.reactivex.exceptions.Exceptions;
+import io.reactivex.exceptions.*;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.subscribers.SinglePostCompleteSubscriber;
@@ -87,7 +87,7 @@ public void onError(Throwable t) {
                 p = ObjectHelper.requireNonNull(onErrorMapper.apply(t), "The onError publisher returned is null");
             } catch (Throwable e) {
                 Exceptions.throwIfFatal(e);
-                actual.onError(e);
+                actual.onError(new CompositeException(t, e));
                 return;
             }
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableUsing.java

@@ -21,6 +21,7 @@
 import io.reactivex.*;
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
@@ -54,7 +55,7 @@ public void subscribeActual(Subscriber<? super T> s) {
 
         Publisher<? extends T> source;
         try {
-            source = sourceSupplier.apply(resource);
+            source = ObjectHelper.requireNonNull(sourceSupplier.apply(resource), "The sourceSupplier returned a null Publisher");
         } catch (Throwable e) {
             Exceptions.throwIfFatal(e);
             try {

src/main/java/io/reactivex/internal/operators/flowable/FlowableWithLatestFromMany.java

@@ -168,7 +168,7 @@ public void onNext(T t) {
             R v;
 
             try {
-                v = ObjectHelper.requireNonNull(combiner.apply(objects), "combiner returned a null value");
+                v = ObjectHelper.requireNonNull(combiner.apply(objects), "The combiner returned a null value");
             } catch (Throwable ex) {
                 Exceptions.throwIfFatal(ex);
                 cancel();
@@ -297,7 +297,7 @@ public void dispose() {
     final class SingletonArrayFunc implements Function<T, R> {
         @Override
         public R apply(T t) throws Exception {
-            return combiner.apply(new Object[] { t });
+            return ObjectHelper.requireNonNull(combiner.apply(new Object[] { t }), "The combiner returned a null value");
         }
     }
 }

src/main/java/io/reactivex/internal/operators/maybe/MaybeZipArray.java

@@ -192,7 +192,7 @@ public void onComplete() {
     final class SingletonArrayFunc implements Function<T, R> {
         @Override
         public R apply(T t) throws Exception {
-            return zipper.apply(new Object[] { t });
+            return ObjectHelper.requireNonNull(zipper.apply(new Object[] { t }), "The zipper returned a null value");
         }
     }
 }

src/main/java/io/reactivex/internal/operators/maybe/MaybeZipIterable.java

@@ -19,6 +19,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.EmptyDisposable;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.operators.maybe.MaybeZipArray.ZipCoordinator;
 
 public final class MaybeZipIterable<T, R> extends Maybe<R> {
@@ -81,7 +82,7 @@ protected void subscribeActual(MaybeObserver<? super R> observer) {
     final class SingletonArrayFunc implements Function<T, R> {
         @Override
         public R apply(T t) throws Exception {
-            return zipper.apply(new Object[] { t });
+            return ObjectHelper.requireNonNull(zipper.apply(new Object[] { t }), "The zipper returned a null value");
         }
     }
 }

src/main/java/io/reactivex/internal/operators/observable/ObservableInternalHelper.java

@@ -77,7 +77,8 @@ static final class ItemDelayFunction<T, U> implements Function<T, ObservableSour
 
         @Override
         public ObservableSource<T> apply(final T v) throws Exception {
-            return new ObservableTake<U>(itemDelay.apply(v), 1).map(Functions.justFunction(v)).defaultIfEmpty(v);
+            ObservableSource<U> o = ObjectHelper.requireNonNull(itemDelay.apply(v), "The itemDelay returned a null ObservableSource");
+            return new ObservableTake<U>(o, 1).map(Functions.justFunction(v)).defaultIfEmpty(v);
         }
     }
 
@@ -165,7 +166,7 @@ static final class FlatMapWithCombinerOuter<T, R, U> implements Function<T, Obse
         @Override
         public ObservableSource<R> apply(final T t) throws Exception {
             @SuppressWarnings("unchecked")
-            ObservableSource<U> u = (ObservableSource<U>)mapper.apply(t);
+            ObservableSource<U> u = (ObservableSource<U>)ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null ObservableSource");
             return new ObservableMap<U, R>(u, new FlatMapWithCombinerInner<U, R, T>(combiner, t));
         }
     }
@@ -185,7 +186,7 @@ static final class FlatMapIntoIterable<T, U> implements Function<T, ObservableSo
 
         @Override
         public ObservableSource<U> apply(T t) throws Exception {
-            return new ObservableFromIterable<U>(mapper.apply(t));
+            return new ObservableFromIterable<U>(ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Iterable"));
         }
     }
 
@@ -319,7 +320,7 @@ static final class ObservableMapper<T,R> implements Function<T,Observable<R>> {
         @Override
         public Observable<R> apply(T t) throws Exception {
             return RxJavaPlugins.onAssembly(new SingleToObservable<R>(
-                ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null value")));
+                ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null SingleSource")));
         }
 
     }
@@ -403,7 +404,8 @@ static final class ReplayFunction<T, R> implements Function<Observable<T>, Obser
 
         @Override
         public ObservableSource<R> apply(Observable<T> t) throws Exception {
-            return Observable.wrap(selector.apply(t)).observeOn(scheduler);
+            ObservableSource<R> apply = ObjectHelper.requireNonNull(selector.apply(t), "The selector returned a null ObservableSource");
+            return Observable.wrap(apply).observeOn(scheduler);
         }
     }
 }

src/main/java/io/reactivex/internal/operators/observable/ObservableMapNotification.java

@@ -13,14 +13,14 @@
 
 package io.reactivex.internal.operators.observable;
 
-import io.reactivex.internal.functions.ObjectHelper;
 import java.util.concurrent.Callable;
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.exceptions.Exceptions;
+import io.reactivex.exceptions.*;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.functions.ObjectHelper;
 
 public final class ObservableMapNotification<T, R> extends AbstractObservableWithUpstream<T, ObservableSource<? extends R>> {
 
@@ -106,7 +106,7 @@ public void onError(Throwable t) {
                 p = ObjectHelper.requireNonNull(onErrorMapper.apply(t), "The onError ObservableSource returned is null");
             } catch (Throwable e) {
                 Exceptions.throwIfFatal(e);
-                actual.onError(e);
+                actual.onError(new CompositeException(t, e));
                 return;
             }
 

src/main/java/io/reactivex/internal/operators/observable/ObservableReplay.java

@@ -24,6 +24,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
 import io.reactivex.internal.disposables.*;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.fuseable.HasUpstreamObservableSource;
 import io.reactivex.internal.util.*;
 import io.reactivex.observables.ConnectableObservable;
@@ -1026,8 +1027,8 @@ protected void subscribeActual(Observer<? super R> child) {
             ConnectableObservable<U> co;
             ObservableSource<R> observable;
             try {
-                co = connectableFactory.call();
-                observable = selector.apply(co);
+                co = ObjectHelper.requireNonNull(connectableFactory.call(), "The connectableFactory returned a null ConnectableObservable");
+                observable = ObjectHelper.requireNonNull(selector.apply(co), "The selector returned a null ObservableSource");
             } catch (Throwable e) {
                 Exceptions.throwIfFatal(e);
                 EmptyDisposable.error(e, child);