2.x: add ParallelTransformer interface, params-validation

src/main/java/io/reactivex/Completable.java

@@ -1010,7 +1010,7 @@ public final Completable cache() {
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Completable compose(CompletableTransformer transformer) {
-        return wrap(transformer.apply(this));
+        return wrap(ObjectHelper.requireNonNull(transformer, "transformer is null").apply(this));
     }
 
     /**
@@ -1866,7 +1866,7 @@ private Completable timeout0(long timeout, TimeUnit unit, Scheduler scheduler, C
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> U to(Function<? super Completable, U> converter) {
         try {
-            return converter.apply(this);
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             throw ExceptionHelper.wrapOrThrow(ex);

src/main/java/io/reactivex/Flowable.java

@@ -6592,7 +6592,7 @@ public final <U> Single<U> collectInto(final U initialItem, BiConsumer<? super U
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Flowable<R> compose(FlowableTransformer<? super T, ? extends R> composer) {
-        return fromPublisher(((FlowableTransformer<T, R>) composer).apply(this));
+        return fromPublisher(((FlowableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
     }
 
     /**
@@ -14543,7 +14543,7 @@ public final Flowable<Timed<T>> timestamp(final TimeUnit unit, final Scheduler s
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> R to(Function<? super Flowable<T>, R> converter) {
         try {
-            return converter.apply(this);
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             throw ExceptionHelper.wrapOrThrow(ex);

src/main/java/io/reactivex/Maybe.java

@@ -2092,7 +2092,7 @@ public final <U> Maybe<U> cast(final Class<? extends U> clazz) {
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Maybe<R> compose(MaybeTransformer<? super T, ? extends R> transformer) {
-        return wrap(((MaybeTransformer<T, R>) transformer).apply(this));
+        return wrap(((MaybeTransformer<T, R>) ObjectHelper.requireNonNull(transformer, "transformer is null")).apply(this));
     }
 
     /**
@@ -3065,7 +3065,7 @@ public final <U> Maybe<U> ofType(final Class<U> clazz) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> R to(Function<? super Maybe<T>, R> convert) {
         try {
-            return convert.apply(this);
+            return ObjectHelper.requireNonNull(convert, "convert is null").apply(this);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             throw ExceptionHelper.wrapOrThrow(ex);

src/main/java/io/reactivex/Observable.java

@@ -5921,7 +5921,7 @@ public final <U> Single<U> collectInto(final U initialValue, BiConsumer<? super
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Observable<R> compose(ObservableTransformer<? super T, ? extends R> composer) {
-        return wrap(((ObservableTransformer<T, R>) composer).apply(this));
+        return wrap(((ObservableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
     }
 
     /**
@@ -12226,7 +12226,7 @@ public final Observable<Timed<T>> timestamp(final TimeUnit unit, final Scheduler
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> R to(Function<? super Observable<T>, R> converter) {
         try {
-            return converter.apply(this);
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             throw ExceptionHelper.wrapOrThrow(ex);

src/main/java/io/reactivex/Single.java

@@ -1548,7 +1548,7 @@ public final Single<T> hide() {
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Single<R> compose(SingleTransformer<? super T, ? extends R> transformer) {
-        return wrap(((SingleTransformer<T, R>) transformer).apply(this));
+        return wrap(((SingleTransformer<T, R>) ObjectHelper.requireNonNull(transformer, "transformer is null")).apply(this));
     }
 
     /**
@@ -2953,7 +2953,7 @@ private Single<T> timeout0(final long timeout, final TimeUnit unit, final Schedu
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> R to(Function<? super Single<T>, R> convert) {
         try {
-            return convert.apply(this);
+            return ObjectHelper.requireNonNull(convert, "convert is null").apply(this);
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             throw ExceptionHelper.wrapOrThrow(ex);