Fix RxJava 2.X change from Function to ObservableTransformer

Author: pakoito

README.md

@@ -45,7 +45,7 @@ Observable<String> getUserFriends =
 
 ### Compose comprehensions
 
-RxComprehensions contains static methods `doCo()` for `compose()`. Each takes from 1 to 9 `Transformer<T, U>` (RxJava 1.X) or `Function<Observable<T>, Observable<U>` (RxJava 2.X), and returns an `Observable` of the type of the return of the last one.
+RxComprehensions contains static methods `doCo()` for `compose()`. Each takes from 1 to 9 `Transformer<T, U>` (RxJava 1.X) or `ObservableTransformer<T, U>` (RxJava 2.X), and returns an `Observable` of the type of the return of the last one.
 
 ```java
 Observable<List<Siblings>> getRelatives =
@@ -61,15 +61,15 @@ Observable<List<Siblings>> getRelatives =
 
 Observable<RelativeDto> requestRelative(String id) { /* ... */ }
 
-Transformer<RelativeDto, RelativeDto> validate() { /* ... */ }
+ObservableTransformer<RelativeDto, RelativeDto> validate() { /* ... */ }
 
-Transformer<RelativeDto, RelativeDto> assureThreads(Scheduler in, Scheduler out) { /* ... */ }
+ObservableTransformer<RelativeDto, RelativeDto> assureThreads(Scheduler in, Scheduler out) { /* ... */ }
 
-Transformer<RelativeDto, RelativeDto> respectLifecycle(Activity activity) { /* ... */ }
+ObservableTransformer<RelativeDto, RelativeDto> respectLifecycle(Activity activity) { /* ... */ }
 
-Transformer<RelativeDto, Relative> toUILayerModel() { /* ... */ }
+ObservableTransformer<RelativeDto, Relative> toUILayerModel() { /* ... */ }
 
-Transformer<Relative, List<Siblings>> groupSiblings() { /* ... */ }
+ObservableTransformer<Relative, List<Siblings>> groupSiblings() { /* ... */ }
 ```
 
 ## Distribution

rxcomprehensions2/src/main/java/com/pacoworks/rxcomprehensions/RxComprehensions.java

@@ -19,6 +19,7 @@
 import java.util.concurrent.Callable;
 
 import io.reactivex.Observable;
+import io.reactivex.ObservableTransformer;
 import io.reactivex.functions.BiFunction;
 import io.reactivex.functions.Function;
 import io.reactivex.functions.Function3;
@@ -1256,58 +1257,58 @@ public Observable<R> apply(final I i) throws Exception {
 
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<R>> one) throws Exception {
+            final ObservableTransformer<A, R> one) throws Exception {
         return zero.call()
                 .compose(one);
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<R>> two) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, R> two) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two);
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<R>> three) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, R> three) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)
                 .compose(three);
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, D, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<D>> three,
-            final Function<Observable<D>, Observable<R>> four) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, D> three,
+            final ObservableTransformer<D, R> four) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)
@@ -1316,17 +1317,17 @@ public static <A, B, C, D, R> Observable<R> doCo(
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, D, E, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<D>> three,
-            final Function<Observable<D>, Observable<E>> four,
-            final Function<Observable<E>, Observable<R>> five) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, D> three,
+            final ObservableTransformer<D, E> four,
+            final ObservableTransformer<E, R> five) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)
@@ -1336,18 +1337,18 @@ public static <A, B, C, D, E, R> Observable<R> doCo(
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, D, E, F, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<D>> three,
-            final Function<Observable<D>, Observable<E>> four,
-            final Function<Observable<E>, Observable<F>> five,
-            final Function<Observable<F>, Observable<R>> six) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, D> three,
+            final ObservableTransformer<D, E> four,
+            final ObservableTransformer<E, F> five,
+            final ObservableTransformer<F, R> six) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)
@@ -1358,19 +1359,19 @@ public static <A, B, C, D, E, F, R> Observable<R> doCo(
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, D, E, F, G, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<D>> three,
-            final Function<Observable<D>, Observable<E>> four,
-            final Function<Observable<E>, Observable<F>> five,
-            final Function<Observable<F>, Observable<G>> six,
-            final Function<Observable<G>, Observable<R>> seven) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, D> three,
+            final ObservableTransformer<D, E> four,
+            final ObservableTransformer<E, F> five,
+            final ObservableTransformer<F, G> six,
+            final ObservableTransformer<G, R> seven) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)
@@ -1382,20 +1383,20 @@ public static <A, B, C, D, E, F, G, R> Observable<R> doCo(
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, D, E, F, G, H, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<D>> three,
-            final Function<Observable<D>, Observable<E>> four,
-            final Function<Observable<E>, Observable<F>> five,
-            final Function<Observable<F>, Observable<G>> six,
-            final Function<Observable<G>, Observable<H>> seven,
-            final Function<Observable<H>, Observable<R>> eight) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, D> three,
+            final ObservableTransformer<D, E> four,
+            final ObservableTransformer<E, F> five,
+            final ObservableTransformer<F, G> six,
+            final ObservableTransformer<G, H> seven,
+            final ObservableTransformer<H, R> eight) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)
@@ -1408,21 +1409,21 @@ public static <A, B, C, D, E, F, G, H, R> Observable<R> doCo(
     }
 
     /**
-     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(Function)}.
+     * Composes an {@link Observable} from multiple {@link Function} chained by {@link Observable#compose(ObservableTransformer)}.
      *
      * @return composed Observable
      */
     public static <A, B, C, D, E, F, G, H, I, R> Observable<R> doCo(
             final Callable<Observable<A>> zero,
-            final Function<Observable<A>, Observable<B>> one,
-            final Function<Observable<B>, Observable<C>> two,
-            final Function<Observable<C>, Observable<D>> three,
-            final Function<Observable<D>, Observable<E>> four,
-            final Function<Observable<E>, Observable<F>> five,
-            final Function<Observable<F>, Observable<G>> six,
-            final Function<Observable<G>, Observable<H>> seven,
-            final Function<Observable<H>, Observable<I>> eight,
-            final Function<Observable<I>, Observable<R>> nine) throws Exception {
+            final ObservableTransformer<A, B> one,
+            final ObservableTransformer<B, C> two,
+            final ObservableTransformer<C, D> three,
+            final ObservableTransformer<D, E> four,
+            final ObservableTransformer<E, F> five,
+            final ObservableTransformer<F, G> six,
+            final ObservableTransformer<G, H> seven,
+            final ObservableTransformer<H, I> eight,
+            final ObservableTransformer<I, R> nine) throws Exception {
         return zero.call()
                 .compose(one)
                 .compose(two)

rxcomprehensions2/src/test/java/com/pacoworks/rxcomprehensions/RxComprehensionsTest.java

@@ -22,6 +22,8 @@
 import java.util.concurrent.Callable;
 
 import io.reactivex.Observable;
+import io.reactivex.ObservableSource;
+import io.reactivex.ObservableTransformer;
 import io.reactivex.functions.BiFunction;
 import io.reactivex.functions.Function;
 import io.reactivex.functions.Function3;
@@ -322,11 +324,11 @@ public Observable<Object> apply(Integer integer, Boolean aBoolean, String s, Str
         };
     }
 
-    private Function<Observable<Integer>, Observable<String>> intIncrementToString() {
-        return new Function<Observable<Integer>, Observable<String>>() {
+    private ObservableTransformer<Integer, String> intIncrementToString() {
+        return new ObservableTransformer<Integer, String>() {
             @Override
-            public Observable<String> apply(Observable<Integer> integerObservable) throws Exception {
-                return integerObservable.map(new Function<Integer, String>() {
+            public ObservableSource<String> apply(Observable<Integer> upstream) {
+                return upstream.map(new Function<Integer, String>() {
                     @Override
                     public String apply(Integer integer) throws Exception {
                         return Integer.toString(integer + 1);
@@ -336,11 +338,11 @@ public String apply(Integer integer) throws Exception {
         };
     }
 
-    private Function<Observable<String>, Observable<Integer>> stringToInt() {
-        return new Function<Observable<String>, Observable<Integer>>() {
+    private ObservableTransformer<String, Integer> stringToInt() {
+        return new ObservableTransformer<String, Integer>() {
             @Override
-            public Observable<Integer> apply(Observable<String> stringObservable) throws Exception {
-                return stringObservable.map(new Function<String, Integer>() {
+            public ObservableSource<Integer> apply(Observable<String> upstream) {
+                return upstream.map(new Function<String, Integer>() {
                     @Override
                     public Integer apply(String s) throws Exception {
                         return Integer.parseInt(s);