Improve compose() generics

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

@@ -6483,11 +6483,12 @@ public final <U> Single<U> collectInto(final U initialItem, BiConsumer<? super U
      * @return the source Publisher, transformed by the transformer function
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      */
+    @SuppressWarnings("unchecked")
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final <R> Flowable<R> compose(FlowableTransformer<T, R> composer) {
-        return fromPublisher(composer.apply(this));
+    public final <R> Flowable<R> compose(FlowableTransformer<? super T, ? extends R> composer) {
+        return fromPublisher(((FlowableTransformer<T, R>) composer).apply(this));
     }
 
     /**

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

@@ -2074,10 +2074,11 @@ public final <U> Maybe<U> cast(final Class<? extends U> clazz) {
      * @return a Maybe, transformed by the transformer function
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      */
+    @SuppressWarnings("unchecked")
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final <R> Maybe<R> compose(MaybeTransformer<T, R> transformer) {
-        return wrap(transformer.apply(this));
+    public final <R> Maybe<R> compose(MaybeTransformer<? super T, ? extends R> transformer) {
+        return wrap(((MaybeTransformer<T, R>) transformer).apply(this));
     }
 
     /**

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

@@ -5722,10 +5722,11 @@ public final <U> Single<U> collectInto(final U initialValue, BiConsumer<? super
      * @return the source ObservableSource, transformed by the transformer function
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      */
+    @SuppressWarnings("unchecked")
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final <R> Observable<R> compose(ObservableTransformer<T, R> composer) {
-        return wrap(composer.apply(this));
+    public final <R> Observable<R> compose(ObservableTransformer<? super T, ? extends R> composer) {
+        return wrap(((ObservableTransformer<T, R>) composer).apply(this));
     }
 
     /**

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

@@ -1540,10 +1540,11 @@ public final Single<T> hide() {
      * @return the source Single, transformed by the transformer function
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      */
+    @SuppressWarnings("unchecked")
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final <R> Single<R> compose(SingleTransformer<T, R> transformer) {
-        return wrap(transformer.apply(this));
+    public final <R> Single<R> compose(SingleTransformer<? super T, ? extends R> transformer) {
+        return wrap(((SingleTransformer<T, R>) transformer).apply(this));
     }
 
     /**

src/test/java/io/reactivex/TransformerTest.java

@@ -13,12 +13,11 @@
 
 package io.reactivex;
 
-import static org.junit.Assert.*;
-
+import io.reactivex.exceptions.TestException;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 
-import io.reactivex.exceptions.TestException;
+import static org.junit.Assert.*;
 
 public class TransformerTest {
 
@@ -83,7 +82,7 @@ public Maybe<Integer> apply(Maybe<Integer> v) {
     }
 
     @Test
-    public void completabeTransformerThrows() {
+    public void completableTransformerThrows() {
         try {
             Completable.complete().compose(new CompletableTransformer() {
                 @Override
@@ -96,4 +95,73 @@ public Completable apply(Completable v) {
             assertEquals("Forced failure", ex.getMessage());
         }
     }
+
+    // Test demos for signature generics in compose() methods. Just needs to compile.
+
+    @Test
+    public void observableGenericsSignatureTest() {
+        A<String, Integer> a = new A<String, Integer>() { };
+
+        Observable.just(a).compose(TransformerTest.<String>testObservableTransformerCreator());
+    }
+
+    @Test
+    public void singleGenericsSignatureTest() {
+        A<String, Integer> a = new A<String, Integer>() { };
+
+        Single.just(a).compose(TransformerTest.<String>testSingleTransformerCreator());
+    }
+
+    @Test
+    public void maybeGenericsSignatureTest() {
+        A<String, Integer> a = new A<String, Integer>() { };
+
+        Maybe.just(a).compose(TransformerTest.<String>testMaybeTransformerCreator());
+    }
+
+    @Test
+    public void flowableGenericsSignatureTest() {
+        A<String, Integer> a = new A<String, Integer>() { };
+
+        Flowable.just(a).compose(TransformerTest.<String>testFlowableTransformerCreator());
+    }
+
+    interface A<T, R> {}
+    interface B<T> {}
+
+    private static <T> ObservableTransformer<A<T, ?>, B<T>> testObservableTransformerCreator() {
+        return new ObservableTransformer<A<T, ?>, B<T>>() {
+            @Override
+            public ObservableSource<B<T>> apply(Observable<A<T, ?>> a) {
+                return Observable.empty();
+            }
+        };
+    }
+
+    private static <T> SingleTransformer<A<T, ?>, B<T>> testSingleTransformerCreator() {
+        return new SingleTransformer<A<T, ?>, B<T>>() {
+            @Override
+            public SingleSource<B<T>> apply(Single<A<T, ?>> a) {
+                return Single.never();
+            }
+        };
+    }
+
+    private static <T> MaybeTransformer<A<T, ?>, B<T>> testMaybeTransformerCreator() {
+        return new MaybeTransformer<A<T, ?>, B<T>>() {
+            @Override
+            public MaybeSource<B<T>> apply(Maybe<A<T, ?>> a) {
+                return Maybe.empty();
+            }
+        };
+    }
+
+    private static <T> FlowableTransformer<A<T, ?>, B<T>> testFlowableTransformerCreator() {
+        return new FlowableTransformer<A<T, ?>, B<T>>() {
+            @Override
+            public Publisher<B<T>> apply(Flowable<A<T, ?>> a) {
+                return Flowable.empty();
+            }
+        };
+    }
 }