2.x Remove Function from transformer interfaces to allow a single obj… (#4672)

* 2.x Remove Function from transformer interfaces to allow a single object to implement multiple transformers

* Fix missing Exception declarations from transformers

Author: dalewking

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

@@ -942,7 +942,12 @@ public final Throwable blockingGet(long timeout, TimeUnit unit) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Completable compose(CompletableTransformer transformer) {
-        return wrap(to(transformer));
+        try {
+            return wrap(transformer.apply(this));
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
     }
 
     /**

src/main/java/io/reactivex/CompletableTransformer.java

@@ -13,12 +13,10 @@
 
 package io.reactivex;
 
-import io.reactivex.functions.Function;
-
 /**
  * Convenience interface and callback used by the compose operator to turn a Completable into another
  * Completable fluently.
  */
-public interface CompletableTransformer extends Function<Completable, CompletableSource> {
-
+public interface CompletableTransformer {
+    CompletableSource apply(Completable completable) throws Exception;
 }

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

@@ -6278,7 +6278,12 @@ 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<T, R> composer) {
-        return fromPublisher(to(composer));
+        try {
+            return fromPublisher(composer.apply(this));
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
     }
 
     /**

src/main/java/io/reactivex/FlowableTransformer.java

@@ -15,14 +15,12 @@
 
 import org.reactivestreams.Publisher;
 
-import io.reactivex.functions.Function;
-
 /**
  * Interface to compose Flowables.
  *
  * @param <Upstream> the upstream value type
  * @param <Downstream> the downstream value type
  */
-public interface FlowableTransformer<Upstream, Downstream> extends Function<Flowable<Upstream>, Publisher<? extends Downstream>> {
-
+public interface FlowableTransformer<Upstream, Downstream> {
+    Publisher<? extends Downstream> apply(Flowable<Upstream> flowable) throws Exception;
 }

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

@@ -1974,7 +1974,12 @@ public final <U> Maybe<U> cast(final Class<? extends U> clazz) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Maybe<R> compose(MaybeTransformer<T, R> transformer) {
-        return wrap(to(transformer));
+        try {
+            return wrap(transformer.apply(this));
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
     }
 
     /**

src/main/java/io/reactivex/MaybeTransformer.java

@@ -13,14 +13,12 @@
 
 package io.reactivex;
 
-import io.reactivex.functions.Function;
-
 /**
  * Interface to compose Maybes.
  *
  * @param <Upstream> the upstream value type
  * @param <Downstream> the downstream value type
  */
-public interface MaybeTransformer<Upstream, Downstream> extends Function<Maybe<Upstream>, MaybeSource<Downstream>> {
-
+public interface MaybeTransformer<Upstream, Downstream> {
+    MaybeSource<Downstream> apply(Maybe<Upstream> maybe) throws Exception;
 }

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

@@ -5489,7 +5489,12 @@ public final <U> Single<U> collectInto(final U initialValue, BiConsumer<? super
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Observable<R> compose(ObservableTransformer<T, R> composer) {
-        return wrap(to(composer));
+        try {
+            return wrap(composer.apply(this));
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
     }
 
 

src/main/java/io/reactivex/ObservableTransformer.java

@@ -13,14 +13,12 @@
 
 package io.reactivex;
 
-import io.reactivex.functions.Function;
-
 /**
  * Interface to compose Observables.
  *
  * @param <Upstream> the upstream value type
  * @param <Downstream> the downstream value type
  */
-public interface ObservableTransformer<Upstream, Downstream> extends Function<Observable<Upstream>, ObservableSource<Downstream>> {
-
+public interface ObservableTransformer<Upstream, Downstream> {
+    ObservableSource<Downstream> apply(Observable<Upstream> upstream) throws Exception;
 }

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

@@ -1473,7 +1473,12 @@ public final Single<T> hide() {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Single<R> compose(SingleTransformer<T, R> transformer) {
-        return wrap(to(transformer));
+        try {
+            return wrap(transformer.apply(this));
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
     }
 
     /**

src/main/java/io/reactivex/SingleTransformer.java

@@ -13,14 +13,12 @@
 
 package io.reactivex;
 
-import io.reactivex.functions.Function;
-
 /**
  * Interface to compose Singles.
  *
  * @param <Upstream> the upstream value type
  * @param <Downstream> the downstream value type
  */
-public interface SingleTransformer<Upstream, Downstream> extends Function<Single<Upstream>, SingleSource<Downstream>> {
-
+public interface SingleTransformer<Upstream, Downstream> {
+    SingleSource<Downstream> apply(Single<Upstream> upstream) throws Exception;
 }