Fix the compose generics

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/JavaConversions.scala

@@ -65,8 +65,8 @@ object JavaConversions {
 
   implicit def toJavaTransformer[T, R](transformer: Observable[T] => Observable[R]): rx.Observable.Transformer[T, R] = {
     new rx.Observable.Transformer[T, R] {
-      override def call(o: rx.Observable[_ <: T]): rx.Observable[R] = {
-        transformer(toScalaObservable(o)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+      override def call(o: rx.Observable[_ <: T]): rx.Observable[_ <: R] = {
+        transformer(toScalaObservable(o)).asJavaObservable
       }
     }
   }

rxjava-core/src/main/java/rx/Observable.java

@@ -204,16 +204,18 @@ public void call(Subscriber<? super R> o) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      * @since 0.20
      */
-    public <R> Observable<R> compose(Transformer<? super T, R> transformer) {
-        return transformer.call(this);
+    @SuppressWarnings("unchecked")
+    public <R> Observable<R> compose(Transformer<? super T, ? extends R> transformer) {
+        // Casting to Observable<R> is type-safe because we know Observable is covariant.
+        return (Observable<R>) transformer.call(this);
     }
 
     /**
      * Transformer function used by {@link #compose}.
      * @warn more complete description needed
      * @since 0.20
      */
-    public static interface Transformer<T, R> extends Func1<Observable<? extends T>, Observable<R>> {
+    public static interface Transformer<T, R> extends Func1<Observable<? extends T>, Observable<? extends R>> {
         // cover for generics insanity
     }
 

rxjava-core/src/test/java/rx/CovarianceTest.java

@@ -20,6 +20,7 @@
 import org.junit.Test;
 
 import rx.Observable.Transformer;
+import rx.functions.Func1;
 import rx.functions.Func2;
 
 /**
@@ -61,28 +62,61 @@ public Integer call(Media t1, Media t2) {
 
     @Test
     public void testCovarianceOfCompose() {
-        Observable<HorrorMovie> movie = Observable.<HorrorMovie> from(new HorrorMovie());
+        Observable<HorrorMovie> movie = Observable.just(new HorrorMovie());
         Observable<Movie> movie2 = movie.compose(new Transformer<Movie, Movie>() {
 
             @Override
-            public Observable<Movie> call(Observable<? extends Movie> t1) {
-                return Observable.from(new Movie());
+            public Observable<? extends Movie> call(Observable<? extends Movie> t1) {
+                return Observable.just(new Movie());
             }
 
         });
     }
 
     @Test
     public void testCovarianceOfCompose2() {
-        Observable<Movie> movie = Observable.<Movie> from(new HorrorMovie());
+        Observable<Movie> movie = Observable.<Movie> just(new HorrorMovie());
         Observable<HorrorMovie> movie2 = movie.compose(new Transformer<Movie, HorrorMovie>() {
             @Override
-            public Observable<HorrorMovie> call(Observable<? extends Movie> t1) {
-                return Observable.from(new HorrorMovie());
+            public Observable<? extends HorrorMovie> call(Observable<? extends Movie> t1) {
+                return Observable.just(new HorrorMovie());
+            }
+        });
+    }
+
+    @Test
+    public void testCovarianceOfCompose3() {
+        Observable<Movie> movie = Observable.<Movie>just(new HorrorMovie());
+        Observable<HorrorMovie> movie2 = movie.compose(new Transformer<Movie, HorrorMovie>() {
+            @Override
+            public Observable<? extends HorrorMovie> call(Observable<? extends Movie> t1) {
+                return Observable.just(new HorrorMovie()).map(new Func1<HorrorMovie, HorrorMovie>() {
+
+                    @Override
+                    public HorrorMovie call(HorrorMovie horrorMovie) {
+                        return horrorMovie;
+                    }
+                });
+            }
+        });
+    }
+
+    @Test
+    public void testCovarianceOfCompose4() {
+        Observable<HorrorMovie> movie = Observable.just(new HorrorMovie());
+        Observable<HorrorMovie> movie2 = movie.compose(new Transformer<HorrorMovie, HorrorMovie>() {
+            @Override
+            public Observable<? extends HorrorMovie> call(Observable<? extends HorrorMovie> t1) {
+                return t1.map(new Func1<HorrorMovie, HorrorMovie>() {
+
+                    @Override
+                    public HorrorMovie call(HorrorMovie horrorMovie) {
+                        return horrorMovie;
+                    }
+                });
             }
         });
     }
-
 
     /*
      * Most tests are moved into their applicable classes such as [Operator]Tests.java