Gradle | #iOS | #Android ファショ通

直感的に理解する RxJava その2: Operator

最終更新日時 : 2016-04-05 06:39

ReactiveX

その1では, RxJava の基本的な構造と map() オペレータの紹介をしました. しかし, これだけではまだまだ実際には利用できません.

しかし, RxJava の実力の大部分はオペレータなので, すぐに利用できるようになることができます.

サンプルをみながらさらに Operator を学んでいきまよう.

設定

利用可能なこのメソッドがあるとします.


// テキスト検索からWEBサイトのURLの List を返す
Observable<List<String>> query(String text);

テキストで検索してこれらの結果を表示したいとします. 前回の記事をふまえて考えると以下のようなものが思いつくかもしれません.


query("Hello, world!")
    .subscribe(urls -> {
        for (String url : urls) {
            System.out.println(url);
        }
    });

これは少し残念です. なぜなら, データストリームの変換を利用していません. もし, それぞれのURLを変形したい場合はすべてを Subscriber の中でやらなければなりません. 便利な map() を利用していません.

urls -> urls 内で map() を作成することもできますが, 残念ながら, その後のすべての map() の呼び出しのたびに for-each ループを持つことになります.

この方法ではどうか

Observable.from() はアイテムのコレクションを取得し, それぞれを一度に発することができます.


Observable.from("url1", "url2", "url3")
    .subscribe(url -> System.out.println(url));

こんな感じで使えます.


query("Hello, world!")
    .subscribe(urls -> {
        Observable.from(urls)
            .subscribe(url -> System.out.println(url));
    });

for-each ループをなくすことはできますが, 結果, コードはややこしくなり入れ子した Subscription になってしまいます. しかも変更しづらく, 実は RxJava の重要な機能を壊しています.

良い方法

ここで flatMap() の登場です.

Observable.flatMap() は Observable から別の Observable を作成します. アイテムからストリームを作成し別の Observable とすることができます.


query("Hello, world!")
    .flatMap(new Func1<List<String>, Observable<String>>() {
        @Override
        public Observable<String> call(List<String> urls) {
            return Observable.from(urls);
        }
    })
    .subscribe(url -> System.out.println(url));

何が起こっているかわかりやすく書いていますが, ラムダ式で簡単にできますね.


query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .subscribe(url -> System.out.println(url));

ここで大事なことは「返された新しい Observable は Subscriber が利用するもの」ということです. それは, 連続した Observable.from() から返された独立した String たちです.

さらにより良く

flatMap() で欲しい Observable を返すことができましたがまだ十分ではありません.

2つ目の次のメソッドを可能にしなければならないと仮定します.


// サイトのタイトルを返す(もし404なら null を返す)
Observable<String> getTitle(String URL);

URLを表示する代わりに, そのページのタイトルを表示したい場合には問題が有りそうです. メソッドはそれぞれのURLに対してのみ動作して, String ではなくObservable を返します.

しかし, それも flatMap() で簡単に解決できます. Subscriber に到達する前にURLリストが個別のアイテムに分割された後は flatMap() の中でそれぞれの URLに対して getTitle() を使うことができます. .


query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(new Func1<String, Observable<String>>() {
        @Override
        public Observable<String> call(String url) {
            return getTitle(url);
        }
    })
    .subscribe(title -> System.out.println(title));

そしてさらにラムダ式でシンプルにします.


query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .subscribe(title -> System.out.println(title));

複数のメソッドと Observable の返却を同時に組み合わせて行うことができます. すばらしいですね.

それだけでなく, どのように2つのAPIコールを組み合わせてひとつのチェインにいれたか, に注意してみてください. いくつでもAPIコールをいれることができます. 知ってますよね. APIコールの同期を保持して, コールバックを連携させて, データを取得することがどれだけ面倒か. これはすでに, コールバック地獄をスキップしています. そして同じロジックの中にまとめて短いリアクティブなコールになっています.

Oprerator の豊富さ

ここまでで2つの Operator しか使っていませんが, もっとたくさんあります. 他のものを使うとどれだけコードを改善できるでしょう?

URLが 404 のとき getTitle() は null を返します. "null" は表示させたくありませんね. 変更してそれらをフィルタしましょう.


query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .subscribe(title -> System.out.println(title));

fileter() は boolean のチェックを通過したもののみ発します.

そして, 今度は, 5個までの結果だけ表示しましょう.


query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .take(5)
    .subscribe(title -> System.out.println(title));

take() は, 設定された数字を最大の個数として発します.

今度は, それぞれのタイトルを保存しましょう.


query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .take(5)
    .doOnNext(title -> saveTitle(title))
    .subscribe(title -> System.out.println(title));

doOnNext() は, 追加したいふるまいをそれぞれのアイテムが発せられると同時に行います. この場合はタイトルを保存しています.

ここまで, ストリームデータをどれだけ簡単に操作できるかみてきました. 材料をあなたのレシピの合わせて混乱することなく何度も処理できますね.

RxJava には, 大量の Operator がありますが何ができるか参照してみる価値はあると思います.
習得するのに時間がかかるかもしれませんが, 習得するとすぐに使えて役に立つと思います.

カスタムオペレータでさえ書くことができます. この記事では書きませんが, 基本あなたがやりたいと思えばできることでしょう.

それがどうしたの?

なぜこれらの operator を使うのか?

3: 「Operator はデータストリームに対してどんな操作でもできる.」

限度となるのはあなた自身です.

シンプルな Operator のチェインで複雑なロジックを設定することができ, 構成したままの状態でコードを小分けに分割します。

それに加えて, 利用するために変換されたデータはどれだけシンプルになっているか考えてみてください. 最後のサンプルのコードでは, 2つのAPIをコールして, データを操作し, それをディスクに保存しました. しかし, Subscriber はそれが利用するシンプルな Observable しか知りません. 「カプセル化」によりコーディングをより簡単にすることができます.

次回, その3ではRxJavaの他の素晴らしい機能であるエラーハンドリングと並行処理についてです.

[原文] Grokking RxJava, Part 2: Operator, Operator

直感的に理解する RxJava その1: 基本的な構成

直感的に理解する RxJava その3: リアクティブであることのメリット

直感的に理解する RxJava その4: Reactive Android

直感的に理解する RxJava その1: 基本的な構成

最終更新日時 : 2016-04-05 06:38

ReactiveX

最近 Android 開発者の間では RxJava が話題です. ただ最初はとっつきにくいのが難点です. 関数型リアクティブプログラミングは命令型プログラミングの感覚から近づくと難しいかもしれませんが, いったん理解すると素晴らしいです.

ここでは RxJava の特色を説明します. 4回分の記事はあなたを入り口に案内するためのもので, すべてを説明はできませんが RxJava やそれがどのように動くかに興味をもっていただければと思います.

基本的な使い方

リアクティブコードは Observable と Subscriber で構成されています. Observable はアイテムを発し, Subscriber はそれらのアイテムを受取ります.

アイテムの発し方には, パターンがあります, Observable は, 0から複数個のアイテムを発します. そして, すべてが成功して完了するか, エラーになると終了します. Observable が Subscriber.onNext() を繰り返し, Subscriber.onComplete() か Subscriber.onError() で終了します.

これは, 標準的な Observer パターンによく似ていますが, ひとつ大きな違いがあります. Observable は, subscribe されないかぎりアイテムを発しません. 言い換えれば, だれかにリッスンされていなければ, アイテムは発せられません.

Hello, world!

サンプルでどのように動くかみてみましょう. 最初は, 基本的な Observable をつくります.


Observable<String> myObservable = Observable.create(
    new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> sub) {
            sub.onNext("Hello, world!");
            sub.onCompleted();
        }
    }
);

この Observable は "Hello, world!" を発して終了します.

このデータを受けとる Subscriber を作ります.


Subscriber<String> mySubscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) { System.out.println(s); }

    @Override
    public void onCompleted() { }

    @Override
    public void onError(Throwable e) { }
};

これがやっていることは, Observable によって発された文字列を表示するだけです.

Observable と Subscriber ができたので, それらを subscribe() を使ってフックします.


myObservable.subscribe(mySubscriber);

subscribe されたとき Observable は Subscriber の onNext() と onComplete() メソッドをコールします. その結果, Subscriber は "Hello, world!" を出力して終了します.

コードの簡略化

"Hello, world!" と出力させるだけでたくさんのボイラープレートコードがあります. これは何が起きているかすべてみることができるように冗長にしています. RxJavaではコードをシンプルにするためにたくさんのショートカットがあります.

最初に Observable を簡単にしてみましょう. RxJava は一般的な処理に対しての Observable の作成メソッドをいくつか持っています. この場合では Observable.just() を使って上のコードとまったく同じようにひとつのアイテムを発して完了するように書き換えることができます.


Observable<String> myObservable =
    Observable.just("Hello, world!");

次は, Subscriber です. まずは, onCompleted() や onError() を無視して, 代わりに onNext() のふるまいを定義するクラスを使います.


Action1<String> onNextAction = new Action1<String>() {
    @Override
    public void call(String s) {
        System.out.println(s);
    }
};

Action は Subscriber の一部を定義することができます. Observable.subscribe() は, 3つまでの Actionパラメータ onNext(), onError(), and onComplete() を渡すことができ, Subscriber は以下のように書き換えることができます.


myObservable.subscribe(onNextAction, onErrorAction, onCompleteAction);

onError() と OnComplete() は無視するので必要なのは, 一つ目のパラメータだけです.


myObservable.subscribe(onNextAction);

ここで, メソッドチェインを使ってこれらを書き換えてみます.


Observable.just("Hello, world!")
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
              System.out.println(s);
        }
    });

最後に Java8 ラムダ式を使って Action1 のコードの冗長部分を取り除きます.


Observable.just("Hello, world!")
    .subscribe(s -> System.out.println(s));

Android であれば retrolambda の利用を強くお勧めします. 冗長なあなたのコードをおどろくほど短くしてくれます.

変換処理

少し書き換えてみましょう.

署名を "Hello, world!" に加える場合を考えてみます. まず考えられるのは Observable を変更することです.


Observable.just("Hello, world! -Dan")
    .subscribe(s -> System.out.println(s));

Observable をコントロールできる場合であればいいですが, 他人が作ったライブラリを使うことも考えられます. また複数の場所で Observable を利用する場合は何回も書き加えることになります.

代わりに Subscriber を変更したらどうでしょう.


Observable.just("Hello, world!")
    .subscribe(s -> System.out.println(s + " -Dan"));

これはまだ満足できるものではありません. Subscriber はメインスレッドで実行されるかもしれないのでできるだけ軽量にしておくべきです. また, 概念的なレベルで見ると Subscriber は「反応するもの」で「変換するもの」ではありません.

これらの中間で "Hello, world!" を変換できれば良いのではないか?

Operator の紹介

ここでは, Operator を使ってどのように変換処理を行うかを紹介します. Operator は, 元の Observable から最終の Subscriber までの間に, 発されたアイテムを操作します. RxJava にはかなり多くの Operator がありますが, 最初は少しだけ使ってみましょう.

この場合は, 発せられたアイテムを map() で変換することができます:


Observable.just("Hello, world!")
    .map(new Func1<String, String>() {
        @Override
        public String call(String s) {
            return s + " -Dan";
        }
    })
    .subscribe(s -> System.out.println(s));

再度, ラムダ式を使って簡単にします.


Observable.just("Hello, world!")
    .map(s -> s + " -Dan")
    .subscribe(s -> System.out.println(s));

かなりクールではないですか?

map() は基本的なアイテム変換の Operator です. 最終の Subscriber が使える形にまとめたり, データを完全に適合させるために, 多くの map() をチェインすることができます.

さらに map() を

map() には面白い特徴があります. 元 Observable と同じタイプのアイテムを発する必要はありません.

Subscriber がオリジナルのテキストを出力する代わりにテキストのハッシュを出力したいならば以下のようになります.


Observable.just("Hello, world!")
    .map(new Func1<String, Integer>() {
        @Override
        public Integer call(String s) {
            return s.hashCode();
        }
    })
    .subscribe(i -> System.out.println(Integer.toString(i)));

おもしろいですね.
String から始まったのに Subscriber は Integer を受け取ります. このコードをラムダ式で短くしましょう.


Observable.just("Hello, world!")
    .map(s -> s.hashCode())
    .subscribe(i -> System.out.println(Integer.toString(i)));

最初に言ったように, Subscriber はできるだけ短くしたいので, もうひとつ map() を使って String に戻しましょう.


Observable.just("Hello, world!")
    .map(s -> s.hashCode())
    .map(i -> Integer.toString(i))
    .subscribe(s -> System.out.println(s));

元のコードの件に戻ってみましょう. いくつかの変換のステップを追加するだけです。同様に署名変換を追加することができます。


Observable.just("Hello, world!")
    .map(s -> s + " -Dan")
    .map(s -> s.hashCode())
    .map(i -> Integer.toString(i))
    .subscribe(s -> System.out.println(s));

だから何?

ここであなたは思ったかもしれません.

「簡単なコードなのにいろいろやりすぎ」

確かにそうです.

これらは簡単なコードの例でしたが, 大事な2つの考え方が含まれています.

1:「Observable と Subscriber でいろんなことができる」

想像力を働かせてみましょう. いろんなことができます.

Observable はデータベースクエリーとなることができます. Subscriber は結果を受け取り画面に表示します. また, Observable は画面のクリックにもなることもでき, Subscriber ではそれに反応できます. あるいは, Observable はインターネットから取得したバイトストリームになり, Subscriber ではそれをディスクに書き込むことができます.

どんな処理もできる一般的なフレームワークなのです.

2:「Observable と Subscriber はそれらの間にある変換処理から独立している」

最初の Observable と最後の Subscriber の間の map() 処理に集中して取り組む事ができます.
高度に組み合わせが可能で, データ操作しやすく, 入出力データが正しく Operator で操作できるようにいくらでもメソッドをチェインすることができます.

これら 2つのキーとなる考え方を組み合わせることでたくさんの潜在能力を引き出すことができます. ここでは, map() ひとつだけを利用しましたがこれだけでは限界があります. その2では RxJava で利用できる多くの Operator を利用してみます.

[原文] Grokking RxJava, Part 1: The Basics

直感的に理解する RxJava その2: Operator

直感的に理解する RxJava その3: リアクティブであることのメリット

直感的に理解する RxJava その4: Reactive Android