Android rxjava flapMap example

Include the rxjava and rxandroid in the dependencies in your gradle build file.

compile 'io.reactivex:rxjava:1.1.5'
compile 'io.reactivex:rxandroid:1.2.0'

This is a method mocking a long running task.

private Observable<Integer> multiplyInt(final Integer integer, final int mulplier) {
    for (int i=0; i<1000000000; i++) {}
    return Observable.just(new Integer(integer * mulplier));
}

In the 2 examples below, subscribeOn(Schedulers.io()) means make it to run in a background thread Schedulers.io, observeOn(AndroidSchedulers.mainThread()) means when the tasks are completed, run the result in main thread AndroidSchedulers.mainThread(), and subscribe means run it, similar to execute or run, basically to trigger the task to run.

This rxjava flapMap example demonstrates using flapMap to pass the result from one to another. In this case, the initial input is 2, then it’s being multiplied by 2 and the result is passed to the next flapMap, then gets multiplied by 3 and the result is passed to the next flapMap, then gets multiplied by 4 and the result is passed down to the subscribe and the final number 48 is being printed in the onNext() method in the subscribe.

// Input 2
// 1st flapMap  2 * 2
// 2nd flapMap 4 * 3
// 3nd flapMap 12 * 4
// result in the onNext() method in the subscribe is 48
private void flapMapEx1() {
    Observable.just(2)
            .flatMap(new Func1<Integer, Observable<Integer>>() {
                @Override
                public Observable<Integer> call(Integer integer) {
                    Log.d("FlapmapActivity", integer + " * 2");
                    return multiplyInt(integer, 2);
                }
            })
            .flatMap(new Func1<Integer, Observable<Integer>>() {
                @Override
                public Observable<Integer> call(Integer integer) {
                    Log.d("FlapmapActivity", integer + " * 3");
                    return multiplyInt(integer, 3);
                }
            })
            .flatMap(new Func1<Integer, Observable<Integer>>() {
                @Override
                public Observable<Integer> call(Integer integer) {
                    Log.d("FlapmapActivity", integer + " * 4");
                    return multiplyInt(integer, 4);
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Integer>() {
                @Override
                public void onCompleted() {
                    Log.d("FlapmapActivity", "flapMapEx onCompleted");
                }

                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                    Log.d("FlapmapActivity", "flapMapEx onError");
                }

                @Override
                public void onNext(Integer integer) {
                    Log.d("FlapmapActivity", "flapMapEx " + integer.toString());
                }
            });
}

This example demonstrates the use of flapMap, Observable.from(), and filter in rxjava. The initial input is a list of integers from 1 to 9. In the first flapMap, each Integer is being passed down to the filter one by one through Observable.from(). Then the filter ignores the odd integers, and pass down the even integer one at a time to the next flapMap. The final flapMap multiply each integer it received by 2, and emits to the observer in the subscribe. Lastly, each of the Integer from the input that is an even integer is doubled and printed in the onNext() method one by one.

// 1,2,3,4,5,6,7,8,9
// filter out odd numbers 1, 3, 5, 6, 9
// pass down the even numbers 2, 4, 6, 8 to the last flatMap
// the last flapMap multiply each even number by 2
// the onNext in the subscribe prints 4, 8, 12, 16 one by one
private void flapMapEx2() {
    List<Integer> ints = new ArrayList<>();
    for (int i=1; i<10; i++) {
        ints.add(new Integer(i));
    }

    Log.d("Rx8FlapmapActivity", "flapMapEx 1,2,3,4,5,6,7,8,9");

    Observable.just(ints)
            .flatMap(new Func1<List<Integer>, Observable<Integer>>() {
                @Override
                public Observable<Integer> call(List<Integer> ints) {
                    return Observable.from(ints);
                }
            })
            .filter(new Func1<Integer, Boolean>() {
                @Override
                public Boolean call(Integer integer) {
                    Log.d("Rx8FlapmapActivity", "flapMapEx2 filter out odd numbers.........");
                    return integer.intValue() % 2 == 0;
                }
            })
            .flatMap(new Func1<Integer, Observable<Integer>>() {
                @Override
                public Observable<Integer> call(Integer integer) {
                    for (int i = 0; i < 1000000000; i++) {
                    }
                    return multiplyInt(integer, 2);
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Integer>() {
                @Override
                public void onCompleted() {
                    Log.d("FlapmapActivity", "flapMapEx2 onCompleted");
                }

                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                    Log.d("FlapmapActivity", "flapMapEx2 onError");
                }

                @Override
                public void onNext(Integer integer) {
                    Log.d("FlapmapActivity", "flapMapEx2 onNext>>>" + integer.toString());
                }
            });
}