diff --git a/src/main/java/io/reactivex/Completable.java b/src/main/java/io/reactivex/Completable.java
index 6d64f5f8f3..032fa1608f 100644
--- a/src/main/java/io/reactivex/Completable.java
+++ b/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);
+        }
     }
 
     /**
diff --git a/src/main/java/io/reactivex/CompletableTransformer.java b/src/main/java/io/reactivex/CompletableTransformer.java
index fd3204a704..432c699771 100644
--- a/src/main/java/io/reactivex/CompletableTransformer.java
+++ b/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;
 }
diff --git a/src/main/java/io/reactivex/Flowable.java b/src/main/java/io/reactivex/Flowable.java
index a593c84c71..0031f53f4d 100644
--- a/src/main/java/io/reactivex/Flowable.java
+++ b/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);
+        }
     }
 
     /**
diff --git a/src/main/java/io/reactivex/FlowableTransformer.java b/src/main/java/io/reactivex/FlowableTransformer.java
index 825c0511f1..0f2017a02b 100644
--- a/src/main/java/io/reactivex/FlowableTransformer.java
+++ b/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;
 }
diff --git a/src/main/java/io/reactivex/Maybe.java b/src/main/java/io/reactivex/Maybe.java
index b4b559851a..3c3386a6fd 100644
--- a/src/main/java/io/reactivex/Maybe.java
+++ b/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);
+        }
     }
 
     /**
diff --git a/src/main/java/io/reactivex/MaybeTransformer.java b/src/main/java/io/reactivex/MaybeTransformer.java
index ecd14838e5..c169e2023f 100644
--- a/src/main/java/io/reactivex/MaybeTransformer.java
+++ b/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;
 }
diff --git a/src/main/java/io/reactivex/Observable.java b/src/main/java/io/reactivex/Observable.java
index 910e14fdb1..2ed857205f 100644
--- a/src/main/java/io/reactivex/Observable.java
+++ b/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);
+        }
     }
 
 
diff --git a/src/main/java/io/reactivex/ObservableTransformer.java b/src/main/java/io/reactivex/ObservableTransformer.java
index 1a9c192eee..1dec08d247 100644
--- a/src/main/java/io/reactivex/ObservableTransformer.java
+++ b/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;
 }
diff --git a/src/main/java/io/reactivex/Single.java b/src/main/java/io/reactivex/Single.java
index c143e4f148..e367a4baf7 100644
--- a/src/main/java/io/reactivex/Single.java
+++ b/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);
+        }
     }
 
     /**
diff --git a/src/main/java/io/reactivex/SingleTransformer.java b/src/main/java/io/reactivex/SingleTransformer.java
index 5763d034a1..4ce319aec4 100644
--- a/src/main/java/io/reactivex/SingleTransformer.java
+++ b/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;
 }