rxjava-example.java

package net.gini;

import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;

import java.util.*;

/**
 * Created by stephan on 03/01/2017.
 */
public class ReactiveX {

    private static List<String> mCache = null;

    //get cached values
    public static Observable<List<String>> getCachedUserList() {
        return Observable.just(mCache);
    }

    //https://api.github.com/search/users?q=language:kotlin
    public static Observable<List<String>> getUserList() {
        return Observable.just(new ArrayList<>(Arrays.asList("Seymour", "Al", "Ivana", "Hugh", "Yuri", "Kalle", "Hans", "Sepp")));
    }

    //https://api.github.com/users/{username}
    static Observable<String> getEmail(final String user) {
        switch (user) {
            case "Seymour":
                return Observable.just("Seymour@Butts.de");
            case "Al":
                return Observable.just("Al@Koholic.de");
            case "Ivana":
                return Observable.just("Ivana@Tinkle.de");
            case "Hugh":
                return Observable.just("Hugh@Jass.de");
            case "Yuri":
                return Observable.just("Yuri@Nating.de");
        }
        return Observable.just(null);
    }

    public static void main(String[] args) {

        Observable.concat(getCachedUserList(), getUserList())
                .takeFirst(new Func1<List<String>, Boolean>() {
                    @Override
                    public Boolean call(List<String> strings) {
                        return strings != null;
                    }
                })
                .doOnNext(new Action1<List<String>>() {
                    @Override
                    public void call(List<String> strings) {
                        mCache = strings;
                    }
                })
                .flatMap(new Func1<List<String>, Observable<String>>() {
                    @Override
                    public Observable<String> call(List<String> strings) {
                        return Observable.from(strings);
                    }
                })
                .flatMap(new Func1<String, Observable<String>>() {
                    @Override
                    public Observable<String> call(String s) {
                        return getEmail(s);
                    }
                })
                .filter(new Func1<String, Boolean>() {
                    @Override
                    public Boolean call(String s) {
                        return s != null;
                    }
                })
                .take(5)
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        System.out.println(s);
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        System.err.println(throwable.getLocalizedMessage());
                    }
                }, new Action0() {
                    @Override
                    public void call() {
                        System.out.println("onComplete");
                    }
                });
    }
}