Fix compose generics

Author: zsxwing

src/main/java/rx/Observable.java

@@ -174,15 +174,17 @@ public void call(Subscriber<? super R> o) {
      * @return the source Observable, 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>
      */
-    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
      */
-    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
     }
 

src/test/java/rx/CovarianceTest.java

@@ -20,12 +20,13 @@
 import org.junit.Test;
 
 import rx.Observable.Transformer;
+import rx.functions.Func1;
 import rx.functions.Func2;
 
 /**
  * Test super/extends of generics.
  * 
- * See https://github.com/ReactiveX/RxJava/pull/331
+ * See https://github.com/Netflix/RxJava/pull/331
  */
 public class CovarianceTest {
 
@@ -61,11 +62,11 @@ public Integer call(Media t1, Media t2) {
 
     @Test
     public void testCovarianceOfCompose() {
-        Observable<HorrorMovie> movie = Observable.<HorrorMovie> just(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) {
+            public Observable<? extends Movie> call(Observable<? extends Movie> t1) {
                 return Observable.just(new Movie());
             }
 
@@ -77,12 +78,45 @@ public void testCovarianceOfCompose2() {
         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) {
+            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