2.x: Observable.compose to use ObservableTransformer (#4658)

Author: akarnokd

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

@@ -5488,7 +5488,7 @@ public final <U> Single<U> collectInto(final U initialValue, BiConsumer<? super
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final <R> Observable<R> compose(Function<? super Observable<T>, ? extends ObservableSource<R>> composer) {
+    public final <R> Observable<R> compose(ObservableTransformer<T, R> composer) {
         return wrap(to(composer));
     }
 

src/test/java/io/reactivex/internal/operators/single/SingleAmbTest.java

@@ -17,9 +17,6 @@
 
 import java.util.*;
 
-import io.reactivex.Completable;
-import io.reactivex.SingleSource;
-import io.reactivex.exceptions.TestException;
 import org.junit.Test;
 
 import io.reactivex.Single;

src/test/java/io/reactivex/observable/ObservableCovarianceTest.java

@@ -90,7 +90,7 @@ public void accept(Movie pv) {
                         System.out.println(pv);
                     }
                 })
-                .compose(new Function<Observable<Movie>, Observable<Movie>>() {
+                .compose(new ObservableTransformer<Movie, Movie>() {
                     @Override
                     public Observable<Movie> apply(Observable<Movie> m) {
                         return m.concatWith(Observable.just(new ActionMovie()));
@@ -116,7 +116,7 @@ public String apply(Movie v) {
     @Test
     public void testCovarianceOfCompose() {
         Observable<HorrorMovie> movie = Observable.just(new HorrorMovie());
-        Observable<Movie> movie2 = movie.compose(new Function<Observable<HorrorMovie>, Observable<Movie>>() {
+        Observable<Movie> movie2 = movie.compose(new ObservableTransformer<HorrorMovie, Movie>() {
             @Override
             public Observable<Movie> apply(Observable<HorrorMovie> t) {
                 return Observable.just(new Movie());
@@ -128,7 +128,7 @@ public Observable<Movie> apply(Observable<HorrorMovie> t) {
     @Test
     public void testCovarianceOfCompose2() {
         Observable<Movie> movie = Observable.<Movie> just(new HorrorMovie());
-        Observable<HorrorMovie> movie2 = movie.compose(new Function<Observable<Movie>, Observable<HorrorMovie>>() {
+        Observable<HorrorMovie> movie2 = movie.compose(new ObservableTransformer<Movie, HorrorMovie>() {
             @Override
             public Observable<HorrorMovie> apply(Observable<Movie> t) {
                 return Observable.just(new HorrorMovie());
@@ -140,7 +140,7 @@ public Observable<HorrorMovie> apply(Observable<Movie> t) {
     @Test
     public void testCovarianceOfCompose3() {
         Observable<Movie> movie = Observable.<Movie>just(new HorrorMovie());
-        Observable<HorrorMovie> movie2 = movie.compose(new Function<Observable<Movie>, Observable<HorrorMovie>>() {
+        Observable<HorrorMovie> movie2 = movie.compose(new ObservableTransformer<Movie, HorrorMovie>() {
             @Override
             public Observable<HorrorMovie> apply(Observable<Movie> t) {
                 return Observable.just(new HorrorMovie()).map(new Function<HorrorMovie, HorrorMovie>() {
@@ -158,7 +158,7 @@ public HorrorMovie apply(HorrorMovie v) {
     @Test
     public void testCovarianceOfCompose4() {
         Observable<HorrorMovie> movie = Observable.just(new HorrorMovie());
-        Observable<HorrorMovie> movie2 = movie.compose(new Function<Observable<HorrorMovie>, Observable<HorrorMovie>>() {
+        Observable<HorrorMovie> movie2 = movie.compose(new ObservableTransformer<HorrorMovie, HorrorMovie>() {
             @Override
             public Observable<HorrorMovie> apply(Observable<HorrorMovie> t1) {
                 return t1.map(new Function<HorrorMovie, HorrorMovie>() {

src/test/java/io/reactivex/observable/ObservableTest.java

@@ -1087,7 +1087,7 @@ public void accept(List<Boolean> booleans) {
     @Test
     public void testCompose() {
         TestObserver<String> ts = new TestObserver<String>();
-        Observable.just(1, 2, 3).compose(new Function<Observable<Integer>, Observable<String>>() {
+        Observable.just(1, 2, 3).compose(new ObservableTransformer<Integer, String>() {
             @Override
             public Observable<String> apply(Observable<Integer> t1) {
                 return t1.map(new Function<Integer, String>() {