RxJavaTest.java

package rxjava;

import com.google.common.base.Stopwatch;
import org.junit.Test;
import org.junit.experimental.runners.Enclosed;
import org.junit.runner.RunWith;
import rx.Notification;
import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static java.lang.String.format;
import static java.util.Arrays.asList;

@RunWith(Enclosed.class)
public class RxJavaTest {

    static final List<Integer> list1 = asList(1, 2, 3, 4, 5);
    static final List<Integer> list2 = asList(6, 7, 8, 9, 10);
    static final List<List<Integer>> list = asList(list1, list2);
    static final Observable<Integer> observable = Observable.from(list1);


    public static class PerformanceCheck {
        void count(String s, Runnable r) {
            System.out.println("[START] " + s + " -----------");
            IntStream.range(1, 10).forEach(p -> {
                Stopwatch sw = Stopwatch.createStarted();
                r.run();
                System.out.println(format("%d Time: %s", p, sw.stop().toString()));
            });
            System.out.println("[END]   " + s + " -----------");
        }

        @Test
        public void Rxjava() throws Exception {

            count("RxJava", () -> {

                List<Integer> res = Observable.from(list1)
                        .filter(i -> i % 2 == 0)
                        .map(i -> {
                            Threads.sleep(10);
                            return i;
                        })
                        .toList()
                        .toBlocking()
                        .single();
            });
        }

        @Test
        public void Stream() throws Exception {

            count("Stream", () -> {

                List<Integer> res = list1.stream()
                        .filter(i -> i % 2 == 0)
                        .map(i -> {
                            Threads.sleep(10);
                            return i;
                        })
                        .collect(Collectors.toList());
            });
        }

    }


    /**
     * 業務ではまだJava7環境も多いのでStream APIも使えなく、
     * また非同期や並列処理も書きにくいのでその辺りを含め記述できる
     * RxJavaでどこまで使えるかいろいろ調べてみました。
     *
     * <p>
     * ここに記載する内容はRxJavaの使い方が中心で、RxJava自体の概念に
     * ついては以下を参考にしてもらえればと思います。
     * http://qiita.com/chooblarin/items/1cb6a58f6e52b8792c52
     * http://techlife.cookpad.com/entry/2015/04/17/100000
     * http://qiita.com/yuya_presto/items/152fa80d073d865cfc90
     * http://qiita.com/yuya_presto/items/c8c3d77ac958c9c8f67b
     *
     *
     *
     * RxJavaを使ううと以下のようなことが機能が実現できます。
     * http://techlife.cookpad.com/entry/2015/04/17/100000
     *
     * List処理の抽象化・ストリーム化
     * Optional
     * Promise
     * Data Binding
     * Event Bus
     *
     * <p>
     * 随時更新予定。
     */
    public static class HowTo {

        /**
         * まずは感覚的に理解しやすいCollectionの操作を行う例。
         * 若干記述量は増えますが、Stream APIのように記述できます。
         */
        @Test
        public void StreamAPIのようにCollection操作ができる() throws Exception {

            List<Integer> res = Observable.from(list) // IterableなオブジェクトからObservableを生成
                    .flatMap(Observable::from)        // Stream APIのflatMap相当
                    .filter(i -> i % 2 == 0)          // Stream APIのfilter相当
                    .map(i -> i * 2)                  // Stream APIのmap相当
                    .skip(1)                          // Stream APIのskip相当
                    .take(3)                          // Stream APIのlimit相当 ※同様の処理のlimit()も存在する
                    .toList()                         // ストリームをListに変換
                    .toBlocking()                     // 同期処理で行う
                    .single();                        // 要素を取得

            System.out.println(res); // [8, 12]
        }

        /**
         * Web APIやDBアクセスなど重い処理を非同期化し合成することができます。
         * ObservableとFuture/Promiseとの違いは、Request/Responseのような一つの要素だけではなく
         * イベントという複数発生する要素をを登録できること。
         * 今回はFutureと同様にObservableに単一のイベントを追加して扱います。
         */
        @Test
        public void Promiseのように扱う() throws Exception {

            // 直列的+並列に処理する
            // 最初にユーザの認証情報を取得し、その認証情報を元に並列にユーザ情報を取得する例
            //
            // getAuthId, ThreadName: RxCachedThreadScheduler-1, Params:id,pass
            // getUserInfo, ThreadName: RxCachedThreadScheduler-2, Params: 123
            // getUserInf, ThreadName: RxCachedThreadScheduler-3, Params: 123
            // Tom has items = [Apple, Banana]
            // All time: 5.301 s
            //
            // 各メソッドの処理時間が1+4+3なので8秒かかるはずの処理が並列に呼び出しを行っているので5秒ですんでいる。
            Stopwatch sw = Stopwatch.createStarted();
            getAuthId("id", "pass")
                    .flatMap(id -> Observable.zip(getUserInfo(id), getUserItems(id),
                            (info, items) -> info + " has items = " + items))
                    .subscribe(new Subscriber<String>() {
                        @Override
                        public void onCompleted() {
                            System.out.println("All time: " + sw.stop());
                        }
                        @Override
                        public void onError(Throwable e) {
                        }
                        @Override
                        public void onNext(String s) {
                            System.out.println(s);
                        }
                    });

            Threads.sleep(7000);
        }

        Observable<Integer> getAuthId(String id, String pass) {
            // 本来以下はHttpアクセスなど
            return Observable.create((Subscriber<? super Integer> subscriber) -> {
                try {
                    System.out.println(String.format("getAuthId, ThreadName: %s, Params:%s,%s", Thread.currentThread().getName(), id, pass));
                    Threads.sleep(1000);
                    subscriber.onNext(123);
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }).subscribeOn(Schedulers.io());
        }

        Observable<String> getUserInfo(Integer authId) {
            // 本来以下はHttpアクセスなど
            return Observable.create((Subscriber<? super String> subscriber) -> {
                try {
                    System.out.println(String.format("getUserInfo, ThreadName: %s, Params: %d", Thread.currentThread().getName(), authId));
                    Threads.sleep(4000);
                    subscriber.onNext("Tom");
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }).subscribeOn(Schedulers.io());
        }

        Observable<List<String>> getUserItems(Integer authId) {
            // 本来以下はHttpアクセスなど
            return Observable.create((Subscriber<? super List<String>> subscriber) -> {
                try {
                    System.out.println(String.format("getUserInf, ThreadName: %s, Params: %d", Thread.currentThread().getName(), authId));
                    Threads.sleep(3000);
                    subscriber.onNext(asList("Apple", "Banana"));
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }).subscribeOn(Schedulers.io());
        }


        /**
         * Java8ならもちろんそのままOptionalを使えばいいのですが、業務要件や
         * Androidなどまだ使えない環境も存在するので、ObservableをOptional化して代用してみます。
         * そのままだと記述すると冗長になるので、今回は {@link rxjava.RxJavaTest.HowTo.Observables} という
         * ユーティリティクラスを作成して使います。
         */
        @Test
        public void Optionalのように扱う() throws Exception {

            Observable<String> exist = Observables.ofNullable("val");
            Observable<Object> empty = Observables.ofNullable(null);

            // Optional#ifPresent
            exist.subscribe(System.out::println); // val
            empty.subscribe(System.out::println); // 表示処理自体も呼ばれない

            // Optional#flatMap, map, filter
            Observables.ofNullable("1,2,3,4")
                    .flatMap(str-> Observable.from(asList(str.split(","))))
                    .map(Integer::parseInt)
                    .filter(i -> i % 2 == 0)
                    .subscribe(System.out::println);

            // Optional#get
            Observables.get(exist); // val
            Observables.get(empty); // throw NoSuchElementException

            // Optional#orElse
            Observables.orElse(exist, "default"); // val
            Observables.orElse(empty, "default"); // default

            // Optional#isPresent
            Observables.isPresent(exist); // true
            Observables.isPresent(empty); // false

        }

        // TODO
        // 直列処理と並列処理の切り替え
        // 同期処理と非同期処理の切り替え


        /**
         * {@link rx.Observable Observable} Utilities.
         */
        static class Observables {

            /**
             * @see java.util.Optional#of(Object)
             */
            static <T> Observable<T> of(T value){
                if(value == null){
                    throw new NullPointerException();
                }
                else{
                    return Observable.just(value);
                }
            }

            /**
             * @see java.util.Optional#ofNullable(Object)
             */
            static <T> Observable<T> ofNullable(T value){
                return value == null ? Observable.empty() : Observable.just(value);
            }

            /**
             * @see java.util.Optional#get()
             */
            static <T> T get(Observable<T> observable){
                return observable.toBlocking().single();
            }
            /**
             * @see java.util.Optional#orElse(Object)
             */
            static <T> T orElse(Observable<T> observable, T other){
                return observable.defaultIfEmpty(other).toBlocking().single();
            }

            /**
             * @see java.util.Optional#isPresent()
             */
            static <T> boolean isPresent(Observable<T> observable){
                return !observable.isEmpty().toBlocking().single();
            }

            /**
             * @see java.util.Optional#orElseGet(Supplier)
             */
            static <T> T orElseGet(Observable<T> observable, Func0<T> other) {
                return isPresent(observable) ? get(observable) : other.call();
            }

            /**
             * @see java.util.Optional#orElseThrow(Supplier)
             */
            static <T, X extends Throwable> T orElseThrow(Observable<T> observable, Func0<? extends X> other) throws X {
                if (isPresent(observable)) {
                    return get(observable);
                } else {
                    throw other.call();
                }
            }
        }

        @Test
        public void subscribeの基本的操作() throws Exception {

            // subscribeがonCompletedした場合
            // Output: 12345>completed!
            observable.subscribe(new Observer<Integer>() {
                @Override
                public void onCompleted() {
                    System.out.println(">completed!");
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println(">" + e.getMessage());
                }

                @Override
                public void onNext(Integer i) {
                    System.out.print(i);
                }
            });

            // subscribeがonErrorした場合
            // Output: 123>error:4
            observable.subscribe(new Observer<Integer>() {
                @Override
                public void onCompleted() {
                    System.out.println(">completed!");
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println(">" + e.getMessage());
                }

                @Override
                public void onNext(Integer i) {
                    if (i == 4) {
                        throw new RuntimeException("error:" + i);
                    }
                    System.out.print(i);
                }
            });
        }


        /**
         * Observableは多くの生成処理が用意されているのでその使い方をまとめます。
         * <p>
         * https://github.com/ReactiveX/RxJava/wiki/Creating-Observables
         */
        @Test
        public void Observableの生成のパターン() throws Exception {

            // IterableなオブジェクトからObservableを生成
            Observable<List<Integer>> o1 = Observable.from(list);

            // 複数のObservableを束ねて生成。flatMapを使う必要がない
            // mergeとconcatの違いはmergeはObservableを束ねる際に全てのストリームの順を考慮して束ねるのに対し、
            // concatは引数で渡されたObservableの順で束ねられる。
            Observable<Integer> o2 = Observable.merge(Observable.from(list1), Observable.from(list2));
            Observable<Integer> o713 = Observable.concat(Observable.from(list1), Observable.from(list2));

            // Subscribeを指定してObservableを生成
            Observable<String> o10 = Observable.create(subscriber -> {
                subscriber.onNext("Hello");
                subscriber.onNext("world!");
                subscriber.onCompleted();
            });

            // 要素の範囲を指定してObservableを生成
            Observable<Integer> o11 = Observable.range(1, 5);

            // 引数で直接Observableを生成
            Observable<Integer> o12 = Observable.just(1, 2, 3);

            // 例外を持ったObservableを生成
            Observable<Object> o111 = Observable.error(new RuntimeException("Error!"));

            // 空のObservableを生成
            Observable<Object> o121 = Observable.empty();

            // TODO
            // zip
            // 渡されたObservableが全て揃った時点で呼び出される。
            // mergeやconcatとの違いは、異なる型のObservableを扱えること。
            //
            // combineLatest
            // zip同様に子どなる方のObservableを扱うが、各要素のonNextが呼ばれた時点でもっとも近いストリームの要素を扱う。


            // cache

            /*
             *
             * Schedulers
             * immediate: 現在のスレッドを指定する。
             * newThread: 新しいスレッドを生成する
             * computation: CPUバウンド用スレッドを生成する。データの加工など向け
             * io: I/Oバウンド用のスレッドを生成する。DBアクセスなど向け
             * http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases
             * http://reactivex.io/RxJava/javadoc/rx/schedulers/Schedulers.html
             *
             */

        }

        /**
         * 生成したObservableには様々な操作を行うことができるのでその使い方をまとめます。
         * いわゆるOperators
         * https://github.com/ReactiveX/RxJava/wiki/Alphabetical-List-of-Observable-Operators
         */
        @Test
        public void Observableの変換パターン() throws Exception {

            // Observableのグルーピングを行う
            // Output:
            // key=0,value=[2, 4]
            // key=1,value=[1, 3, 5]
            observable.groupBy(i -> i % 2)
                    .subscribe(grouped -> {
                        grouped.toList().subscribe(list -> System.out.println(format("key=%d,value=%s", grouped.getKey(), list)));
                    });

            // Observableの要素を集約する
            // Output:
            // [1, 2, 3, 4, 5]
            observable.collect(new ArrayList<>(), (li, i) -> li.add(i))
                    .subscribe(System.out::println);

            // 重複を排除する。
            // また重複方法を選択する distinct(Func1 keySelector) もある
            // Output:
            // 12345
            Observable.from(asList(1, 2, 3, 3, 4, 5, 5))
                    .distinct()
                    .subscribe(System.out::print);

            // ストリームを束ねて、新たなストリームを生成する。
            // 束ね方も複数のオーバーロードが存在する。
            // Output:
            // [1, 2][3, 4][5]
            observable.window(2)
                    .subscribe(i -> i.toList().subscribe(System.out::print));


            // ubscribeとobserveのスレッドを指定する。
            // これにより、以下の処理は非同期処理になる。
            // Output:
            // RxNewThreadScheduler-1
            // RxNewThreadScheduler-1
            // RxNewThreadScheduler-1
            // RxNewThreadScheduler-1
            // RxNewThreadScheduler-1
            observable.subscribeOn(Schedulers.newThread())
                    .subscribe(i -> System.out.print(Thread.currentThread().getName()));
            // map: (RxNewThreadScheduler-1)
            // map: (RxNewThreadScheduler-1)
            // map: (RxNewThreadScheduler-1)
            // subscribe: 2 (RxComputationThreadPool-3)
            // subscribe: 4 (RxComputationThreadPool-3)
            // subscribe: 6 (RxComputationThreadPool-3)
            Observable
                    .range(1, 3)
                    .subscribeOn(Schedulers.newThread())
                    .map(i -> {
                        System.out.println("map: (" + Thread.currentThread().getName() + ")");
                        return i * 2;
                    })
                    .observeOn(Schedulers.computation())
                    .subscribe(i ->
                                    System.out.println("subscribe: " + i + " (" + Thread.currentThread().getName() + ")")
                    );




            // ストリームの処理の間にonNextを実行する
            // Output:
            // doOnNext: 1
            // subscribe: 2
            // doOnNext: 2
            // subscribe: 4
            observable.take(2)
                    .doOnNext(i -> System.out.println("doOnNext: " + i))
                    .map(i -> i * 2)
                    .subscribe(i -> System.out.println("subscribe: " + i));

            // 非同期で指定間隔ごとにsubscribeする
            // toBlockingなどしてもブロックされない。
            // takeなどで回数制限を行うことができる。
            // Output: 12345
            final CountDownLatch latch = new CountDownLatch(5);
            Observable
                    .interval(1, TimeUnit.SECONDS)
                    .subscribe(counter -> {
                        latch.countDown();
                        System.out.print(counter);
                    });
            latch.await();


            // 値の集約を行う
            // Output: 6
            observable.take(3)
                    .reduce((sum, value) -> sum + value)
                    .subscribe(System.out::print);

            // 前回の要素も引数に取るmap。畳み込まないreduce
            // Output: 136
            observable.take(3)
                    .scan((sum, value) -> sum + value)
                    .subscribe(System.out::print);


            // 複数のObservableを合成する
            // 2 + 1 = 3
            // 4 + 3 = 7
            // 6 + 5 = 11
            Observable<Integer> o1 = Observable.just(1, 3, 5);
            Observable<Integer> o2 = Observable.just(2, 4, 6);

            Observable
                    .zip(o1, o2, (d1, d2) -> d1 + " + " + d2 + " = " + (d1 + d2))
                    .subscribe(System.out::println);


            // 例外が発生時の処理を行う。
            // 同じ型のObservableを返り値とするObservable#onErrorResumeNextも存在する。
            // 1
            // Error!!
            // -1
            observable.doOnNext(i -> {
                if (i == 2) throw new RuntimeException("Error!!");
            })
                    .onErrorReturn(e -> {
                        System.out.println(e.getMessage());
                        return -1;
                    })
                    .subscribe(System.out::println);


            // タイムアウトを設定し、タイムアウトした場合にはTimeoutExceptionが発生する
            // java.util.concurrent.TimeoutException
            // -1
            Observable.interval(3, TimeUnit.SECONDS)
                    .timeout(1, TimeUnit.SECONDS)
                    .onErrorReturn(e -> {
                        System.out.println(e);
                        return -1L;
                    })
                    .subscribe(System.out::println);

            // 例外が発生しても指定の回数リトライを行う。
            // リトライ処理を細かく指定できるObservable#retryWhenも存在する。
            // 1
            // 2
            // 1
            // 2
            // 3
            // java.lang.RuntimeException: Error!
            //  -1
            observable
                    .doOnNext(i -> {
                        if (new Random().nextInt(3) == 1) {
                            throw new RuntimeException("Error!");
                        }
                    })
                    .retry(1)
                    .onErrorReturn(e -> {
                        System.out.println(e);
                        return -1;
                    })
                    .subscribe(System.out::println);

            // 一定間隔で値をまとめる
            // [0, 1]
            // [2, 3, 4]
            // [5, 6, 7]
            // ...
            Observable.interval(1, TimeUnit.SECONDS)
                    .buffer(3, TimeUnit.SECONDS)
                    .subscribe(System.out::println);
            Threads.sleep(10000);



            // onCompletedされたときに再度subscribeするオペレータ
            // repeat処理を指定する、repeatWhenも存在する。
            // 1355134043
            // 1691330796
            // 1863470304
            Random r = new Random();
            Observable.create(s -> {
                        s.onNext(r.nextInt());
                        s.onCompleted();
                    })
                    .repeat(3)
                    .subscribe(System.out::println);



            // onCompletedされたときに再度subscribeするオペレータ
            // 1355134043
            // 1691330796
            // 1863470304
            Random random = new Random();
            Observable.create(s -> {
                s.onNext(random.nextInt());
                s.onCompleted();
            })
            .repeatWhen(new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
                @Override
                public Observable<?> call(Observable<? extends Notification<?>> observable) {
                    return null;
                }
            })
            .subscribe(System.out::println);



            // Stream APIのparallelの再現
            Func1<Integer, Integer> byDouble = i -> i * 2;
            Func1<Integer, Integer> byTriple = i -> i * 3;
            Func1<Integer, Boolean> isTen    = i -> i % 10 == 0;
            List<Integer> single = Observable.range(1, 10)
                    .window(1)
                    .map(oo -> oo.subscribeOn(Schedulers.computation()))
                    .flatMap(oo -> oo.map(byDouble)
                            .map(byTriple)
                            .filter(isTen))
                    .toList().toBlocking().single();

            System.out.println(single);

            // 指定した間隔でストリームを分割したObservableのListを生成する。
            // 数値の他に時間や関数など複数の引数をとつオーバーロードが存在する。
            // [1, 2][3, 4][5]
            observable.buffer(2).subscribe(System.out::print);


        }


        @Test
        public void s() throws Exception {
            Random r = new Random();
            Observable.create(s -> {
                        s.onNext(r.nextInt());
                        s.onCompleted();
                    })
                    .cache()
                    .subscribe(System.out::println);

            Threads.sleep(3000);

        }


        /**
         * Observableをsubscribeする方法も複数存在するのでその使い方をまとめます。
         */
        @Test
        public void subscribeする() throws Exception {
            // Observableが保持しているオブジェクトを一つ一つに対して処理を行う
            observable.subscribe(System.out::print); // 12345

        }

        /**
         * ObservableのError Handlingについて
         * <p>
         * https://github.com/ReactiveX/RxJava/wiki/Error-Handling
         *
         * @throws Exception
         */
        @Test
        public void エラーハンドリング() throws Exception {
            // Setup
            // Exercise
            // Verify
        }

        // Utility-Operators
        // https://github.com/ReactiveX/RxJava/wiki/Observable-Utility-Operators



        /*
         * 参考
         * RxJavaのObservable<T>でOptional<T>を代行する
         * http://sys1yagi.hatenablog.com/entry/2015/01/26/183000
         * http://techlife.cookpad.com/entry/2015/04/17/100000
         * http://docs.couchbase.com/developer/java-2.0/observables.html
         *
         */


    }

}