ComposeComprehensionsTest.java

import android.support.v4.util.Pair;

import com.michaelflisar.lumberjack.L;
import com.michaelflisar.multiples.Triple;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

/**
 * Created by flisar on 02.03.2017.
 */

public class test
{
    public static void test()
    {
        // Test sums up all previous numbers and adds 1

        // 1) Manual
        test(1);
        test(2);
        test(3);

        // 2) with compose
        testCompose(1);
        testCompose(2);
        testCompose(3);

        // 3) With compose functions
        testComposeBeautiful(1);
        testComposeBeautiful(2);
        testComposeBeautiful(3);
    }

    // -----------------------
    // Simple test function
    // -----------------------
    
    public static void test(int startValue)
    {
        getObs1(startValue)
                .flatMap(new Function<Integer, Observable<Pair<Integer, Integer>>>()
                {
                    @Override
                    public Observable<Pair<Integer, Integer>> apply(@NonNull Integer res1) throws Exception
                    {
                        return Observable.just(res1).zipWith(getObs2(res1), new BiFunction<Integer, Integer, Pair<Integer, Integer>>()
                        {
                            @Override
                            public Pair<Integer, Integer> apply(@NonNull Integer int1, @NonNull Integer int2) throws Exception
                            {
                                return new Pair<Integer, Integer>(int1, int2);
                            }
                        });
                    }
                })
                .flatMap(new Function<Pair<Integer, Integer>, Observable<Triple<Integer, Integer, Integer>>>()
                {
                    @Override
                    public Observable<Triple<Integer, Integer, Integer>> apply(@NonNull Pair<Integer, Integer> res12) throws Exception
                    {
                        return Observable.just(res12).zipWith(getObs3(res12.first, res12.second), new BiFunction<Pair<Integer, Integer>, Integer, Triple<Integer, Integer, Integer>>()
                        {
                            @Override
                            public Triple<Integer, Integer, Integer> apply(@NonNull Pair<Integer, Integer> int12, @NonNull Integer int3) throws Exception
                            {
                                return new Triple<Integer, Integer, Integer>(int12.first, int12.second, int3);
                            }
                        });
                    }
                })
                .subscribe(new Consumer<Triple<Integer, Integer, Integer>>()
                {
                    @Override
                    public void accept(@NonNull Triple<Integer, Integer, Integer> data) throws Exception
                    {
                        L.d("MANUAL Received values: %d | %d | %d", data.first, data.second, data.third);
                    }
                });
    }

    // -----------------------
    // Simple compose functions
    // -----------------------

    public static void testCompose(int startValue)
    {
        getObs1(startValue)
                .compose(new ObservableTransformer<Integer, Pair<Integer, Integer>>()
                {
                    @Override
                    public Observable<Pair<Integer, Integer>> apply(Observable<Integer> obs1)
                    {
                        return obs1.flatMap(new Function<Integer, Observable<Pair<Integer, Integer>>>()
                        {
                            @Override
                            public Observable<Pair<Integer, Integer>> apply(@NonNull Integer res1) throws Exception
                            {
                                return Observable.just(res1).zipWith(getObs2(res1), new BiFunction<Integer, Integer, Pair<Integer, Integer>>()
                                {
                                    @Override
                                    public Pair<Integer, Integer> apply(@NonNull Integer int1, @NonNull Integer int2) throws Exception
                                    {
                                        return new Pair<Integer, Integer>(int1, int2);
                                    }
                                });
                            }
                        });
                    }
                })
                .compose(new ObservableTransformer<Pair<Integer, Integer>, Triple<Integer, Integer, Integer>>()
                 {

                     @Override
                     public ObservableSource<Triple<Integer, Integer, Integer>> apply(Observable<Pair<Integer, Integer>> obs2)
                     {
                         return obs2.flatMap(new Function<Pair<Integer, Integer>, Observable<Triple<Integer, Integer, Integer>>>()
                         {
                             @Override
                             public Observable<Triple<Integer, Integer, Integer>> apply(@NonNull Pair<Integer, Integer> res12) throws Exception
                             {
                                 return Observable.just(res12).zipWith(getObs3(res12.first, res12.second), new BiFunction<Pair<Integer, Integer>, Integer, Triple<Integer, Integer, Integer>>()
                                 {
                                     @Override
                                     public Triple<Integer, Integer, Integer> apply(@NonNull Pair<Integer, Integer> int12, @NonNull Integer int3) throws Exception
                                     {
                                         return new Triple<Integer, Integer, Integer>(int12.first, int12.second, int3);
                                     }
                                 });
                             }
                         });
                     }
                 })
                .subscribe(new Consumer<Triple<Integer, Integer, Integer>>()
                {
                    @Override
                    public void accept(@NonNull Triple<Integer, Integer, Integer> data) throws Exception
                    {
                        L.d("SIMLPE COMPOSE Received values: %d | %d | %d", data.first, data.second, data.third);
                    }
                });;
    }

    // -----------------------
    // BEAUTIFUL compose function
    // -----------------------

    public static void testComposeBeautiful(int startValue)
    {
        getObs1(startValue)
                .compose(doFlatMap1(res1 -> getObs2(res1)))
                .compose(doFlatMap2(res2 -> getObs3(res2.first, res2.second)))
                .subscribe(data -> L.d("COMPOSE BEAUTIFUL Received values: %d | %d | %d", data.first, data.second, data.third));
    }

    // -----------------------
    // BEAUTIFUL compose transformers
    // -----------------------

    public static <A, B> ObservableTransformer<A, Pair<A, B>> doFlatMap1(Function<A, Observable<B>> func)
    {
        return new ObservableTransformer<A, Pair<A, B>>()
        {
            @Override
            public Observable<Pair<A, B>> apply(Observable<A> obaA)
            {
                return obaA.flatMap(new Function<A, Observable<Pair<A, B>>>()
                {
                    @Override
                    public Observable<Pair<A, B>> apply(@NonNull A a) throws Exception
                    {
                        return Observable.just(a).zipWith(func.apply(a), new BiFunction<A, B, Pair<A, B>>()
                        {
                            @Override
                            public Pair<A, B> apply(@NonNull A a, @NonNull B b) throws Exception
                            {
                                return new Pair<A, B>(a, b);
                            }
                        });
                    }
                });
            }
        };
    }

    public static <A, B, C> ObservableTransformer<Pair<A, B>, Triple<A, B, C>> doFlatMap2(Function<Pair<A, B>, Observable<C>> func)
    {
        return new ObservableTransformer<Pair<A, B>, Triple<A, B, C>>()
        {
            @Override
            public ObservableSource<Triple<A, B, C>> apply(Observable<Pair<A, B>> obsAB)
            {
                return obsAB.flatMap(new Function<Pair<A, B>, Observable<Triple<A, B, C>>>()
                {
                    @Override
                    public Observable<Triple<A, B, C>> apply(@NonNull Pair<A, B> ab) throws Exception
                    {
                        return Observable.just(ab).zipWith(func.apply(ab), new BiFunction<Pair<A, B>, C, Triple<A, B, C>>()
                        {
                            @Override
                            public Triple<A, B, C> apply(@NonNull Pair<A, B> ab, @NonNull C c) throws Exception
                            {
                                return new Triple<A, B, C>(ab.first, ab.second, c);
                            }
                        });
                    }
                });
            }
        };
    }
    
    // -----------------------
    // Helper functions
    // -----------------------

    private static Observable<Integer> getObs1(int startValue)
    {
        return Observable.just(startValue);
    }

    private static Observable<Integer> getObs2(Integer result1)
    {
        return Observable.just(result1 + 1);
    }

    private static Observable<Integer> getObs3(Integer result1, Integer result2)
    {
        return Observable.just(result1 + result2 + 1);
    }
}