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

ReactiveX

その1, その2, その3では, RxJava がどのように機能するかを書きました. Androidアプリの開発では, それをどのように利用するのでしょうか. ここでは Android開発者に向けての実践的なことを書いてみようと思います.

RxAndroid

RxAndroid は Android用の RxJavaエクステンションで, より簡単に利用できるようバインディングなどを含んだものです.

まず最初に, AndroidScheduler です. これは, 既存の Android のもつスレッド処理を考慮したスケジューラを提供します. 別に, UIスレッド処理に関係するコードは必要ありません.
AndroidSchedulers.mainThread() を利用するだけです.


retrofitService.getImage(url)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

もし, あなたが自作の Handler を利用しているのであれば HandlerThreadScheduler を利用して スケジューラとそれをリンクしましょう.

次に, AndroidObservable には, Androidライフサイクル内で有効に利用できる機能があります. それは bindActivity() と bindFragment() です. Observable に対してAndroidSchedulers.mainThread() を使うことで自動的に追加でき, Activity や Fragment が終了した場合にはアイテムを発することを停止することができます.


AndroidObservable.bindActivity(this, retrofitService.getImage(url))
    .subscribeOn(Schedulers.io())
    .subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

また, AndroidObservable.fromBroadcast() は便利です. これは, BroadcasrReceiver のように利用することができる Observable を作成できます. 以下が, ネットワーク接続状態が変化したときに通知するコードです.


IntentFilter filter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
AndroidObservable.fromBroadcast(context, filter)
    .subscribe(intent -> handleConnectivityChange(intent));

その他に, View にバインディングできる ViewObserbable が2つあります. あるタイミングで View がクリックされたというイベントを取得したい場合には ViewObserbable.clicks(), TexitView が持っているテキストの変更を監視する ViewObservable.text() があります. 


ViewObservable.clicks(mCardNameEditText, false)
    .subscribe(view -> handleClick(view));

Retrofit

Android用RESTクライアントの有名なライブラリである Retrofit は RxJava をサポートしています.

通常ではコールバックを追加して非同期処理を実装します.


@GET("/user/{id}/photo")
void getUserPhoto(@Path("id") int id, Callback<Photo> cb);

RxJava を利用していれば, これの代わりに Observable を返すことができます.


@GET("/user/{id}/photo")
Observable<Photo> getUserPhoto(@Path("id") int id);

これだけで Observable に変換することができます. データを取得するだけでなく, 変換も同時に行います.

また, Retrofit は Observable に対して, 簡単に複数のRESTコールを合成することができます. 以下のサンプルでは, 写真データとそのメタデータを取得したいとします. zip を利用して取得結果を合成します.


Observable.zip(
    service.getUserPhoto(id),
    service.getPhotoMetadata(id),
    (photo, metadata) -> createPhotoWithData(photo, metadata))
    .subscribe(photoWithData -> showPhoto(photoWithData));

これに似たサンプルをその2 (flatMap() を利用) で紹介しました. 複数のREST処理は RxJava + Retrofit でこれだけ簡単に利用できるのです.

今までのコードや処理に時間のかかるコード

Retrofit は適切な Observable を返すことができることは分かりましたが, その他のライブラリを利用している場合はどうなるでしょうか. Observable に変換するコードが必要となるのでしょうか? すべてを変更せずにそのコードをどのようにして利用することができるのでしょうか ?

ほとんどの場合, そのコードから Observable を作成するには Observable.just() とObservable.from() で十分であると思います.


private Object oldMethod() { ... }

public Observable<Object> newMethod() {
    return Observable.just(oldMethod());
}

oldMethod() が短時間の処理であればこれで問題ないでしょう. しかし, 時間のかかる処理であればどうなるでしょう? Observable.just() の前に oldMethod() が実行されスレッドをブロックしてしまいます.

この問題の対応策として, 時間のかかる処理を defer() でラップするという技を私はいつも利用しています.


private Object slowBlockingMethod() { ... }

public Observable<Object> newMethod() {
    return Observable.defer(() -> Observable.just(slowBlockingMethod()));
}

このようにすると Observable が subscribe されるまでは slowBlockingMethod() から結果を返さないようになります.

ライフサイクル

最もややこしい部分です. Activity のライフサイクルに対しての操作はどうなるでしょう? 2つの問題がいつも登場します.

1.「Configuration が変更されるとき(端末の回転時など) に Subscription が継続している」

Retrofit を利用してRESTコールを行い, 外部から取得したデータを ListView に表示するとします. もし, ユーザが画面を回転させたらどうなるでしょうか? そのリクエストを継続したい場合はどうすればよいのでしょうか?

2. Observable が Context を参照したままでいるので メモリーリークが発生する.

この問題は Context を保持している Subscription が引き起こしています. View と連携しているときはわかりやすいですが, なぜか保持しています. 最終的には, たくさんのメモリーを食いつぶしてしまいます.

残念ながら, それぞれの問題には一発で解決できる方法はありませんが, 分かりやすく理解するためのいくつかのガイドラインがあります.

最初の問題は, RxJava のもつキャッシュ機能で解決できます. それを使うことで同じ Observable が重複して動かないように unsubscribe/resubscribe できます. cache() や replay() を使うことで, たとえ unsubscribe してる場合でさえ 遅延なしのリクエストを継続することができます. このことは, Activity が再作成後の新しい Subscription によりレジュームできることを意味しています.


Observable<Photo> request = service.getUserPhoto(id).cache();
Subscription sub = request.subscribe(photo -> handleUserPhoto(photo));

// ...When the Activity is being recreated...
sub.unsubscribe();

// ...Once the Activity is recreated...
request.subscribe(photo -> handleUserPhoto(photo));

注意点としては, 両方のタイミングで同じキャッシュされたリクエストを利用していることです. 遅延のないコールは一度しか行われません. どこにリクエストを保存するかはあなた次第ですが, ライフサイクルを考慮しながら, ライフサイクルの影響を受けないどこかに保存しなければなりません.

二つ目の問題は, ライフサイクルを考慮しながらの Subscription に対しての適切な unsubscribe で解決できます. これの一般的な方法は CompositeSubscription を使うことですべての Subscription を保持し, onDestroy() や onDestroyView() のタイミングに一括で unsubscribe する方法です.


private CompositeSubscription mCompositeSubscription
    = new CompositeSubscription();

private void doSomething() {
    mCompositeSubscription.add(
        AndroidObservable.bindActivity(this, Observable.just("Hello, World!"))
        .subscribe(s -> System.out.println(s)));
}

@Override
protected void onDestroy() {
    super.onDestroy();

    mCompositeSubscription.unsubscribe();
}

このように Activity/Fragment で CompositeSubscription を持つことで, それぞれを追加し, その後, 一括ですべてを unsubscribe することができます.

注意することは, 一度 CompositeSubscription.unsubscribe() するとそれは再利用できません. 再度利用する場合は 新しい CompositeSubscription を作成して利用しなければなりません.

という形でどちらの問題もコード追加で解決できます. これらのボイラープレートなしに問題を解決できる天才がいつか現れて欲しいです.

あとがき

Android についてはこれだけではありません. RxJava はこれからも更新され, Android にもさらに対応されていくでしょう. みなさんがこれらのことをさらに理解しようとするとき, RxAndroid はまだ開発中で, よいサンプルがまだありません. 私がここで提示したサンプルはこれから一年間は興味深いものになると思います.

その間, RxJava がコーディングを簡単にするだけのものだけでなく, もっとおもしろいものを探します. もしあなたがまだ納得していなければ, いつか私を見つけてください, それについてビールを飲みながら話します.

[原文] Grokking RxJava, Part 4: Reactive Android

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

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

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

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

ReactiveX

RxJava について, その1では「基本的な構造」, その2では「強力な Operator」についてみてきました. しかし, あなたは RxJava のメリットについて納得していないでしょう. ここでは, RxJava フレームワークがもつ他のいくつかのメリットを明らかにしていきます.

エラーハンドリング

ここまで onComplete() と onError() には触れませんでしたが, これらは, Observable がアイテムを発するのを停止したときに, 成功完了なのか, エラー停止したのかを示しています.

Subscriber は onComplete() と onError() をリッスンする機能をもっています. 実際にやってみましょう.


Observable.just("Hello, world!")
    .map(s -> potentialException(s))
    .map(s -> anotherPotentialException(s))
    .subscribe(new Subscriber<String>() {
        @Override
        public void onNext(String s) { System.out.println(s); }

        @Override
        public void onCompleted() { System.out.println("Completed!"); }

        @Override
        public void onError(Throwable e) { System.out.println("Ouch!"); }
    });

potentialException() と anotherPotentialException() の両方が, Exceptionを throw する可能性があるとしましょう.

すべての Observable は一回のコールにつき onComplete() か onError() で終了します. そのとき, プログラム出力は "Completed!" かまたは, Exception が throw されたときは "Ouch!" を 最後に出力します.

これらから分かることがあります.

1.「Exception が throw されたときはいつでも onError() がコールされる」

このことは, エラーハンドリングを極めてシンプルに可能にすることを意味しています. 結果, これだけですべてのエラーをハンドルできます.

2.「Operator は Exception をハンドルする必要がない」

Exception は onError() までスキップするので Observable チェインのどこかで起きた問題をどのように処理するかを Subscriber までは気にしなくて良いということになります.

3.「Subscriber がアイテムの受取りを完了したかどうか把握できる」

タスクが終了するタイミングを知ることはコード上で必要になることがあります. ただ, Observable が終了しない可能性もあります.

このことは, これまでのエラーハンドリングに比べてかなり分かりやすいでしょう. コールバックを使う場合は, それぞれのコールバックでエラーをハンドルしなければならず, 同じコードの繰り返すことになりますし, それぞれのコールバックは, どのようにエラーをハンドルするか, を知らなければなりません. このことは, コールバック部分のコードは密に呼び出し側と連携されているということになります.

RxJava では, Observable は エラー時の処理を知る必要が無く, エラーの状態をハンドルする必要もありません. エラーの場合は処理をスキップして, すべてのエラーハンドリングを Subscriber に置くことができます.

Scheduler

Androidアプリでネットワークリクエストを利用するとします. それが時間がかかるものであれば, 別のスレッドでロードします. そこで問題がおきたりすることが有ります.

マルチスレッドのAndroidアプリは難しく, 正しいスレッドで正しいコードを実行することを確実に行わなければなりません. これは混乱してアプリがクラッシュする可能性がとなります. メインスレッド以外からViewを操作するとException が発生することはご存知だと思います.

RxJava では, subscribeOn() を使うことで どのスレッド上で実行するかを Observer に教えることができます. また observeOn() を使うことで Subscriber を実行するスレッドを指定できます.


myObservableServices.retrieveImage(url)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

シンプルですね. Subscriber までのすべての処理は I/Oスレッドで実行されます. そして最後にメインスレッドでViewの操作が行われます.

メインスレッドのブロックをできるだけ少なくするために, Subscriber はできるだけ軽くする.

これらは非常に便利で, すべての Observable に subscribeOn() と ObserveOn() を追加することができます. ただの Operator です. 何の Observable であるか, 前の Operator が何をしているかについては気にする必要はありません. 簡単にスレッドの振り分けを指示することができます.

Observable の処理に時間がかかる場合に, Subscriber がすでにI/Oスレッドで準備できていれば, それを監視する必要がないので subscribeOn() から observeOn() を先延ばしにすることは良い実験になります.

AsyncTask などでは, どの部分を並列処理にするかある程度設計しなければなりませんでしたが, RxJavaを使うと そのまま並列処理と追加するだけでよいのです.

Subscription

他にもまだ説明していないメリットがあります. Observable.subscribe() をコールした時これは Subscription を返します. これは, Observable と Subscriber のリンクです.


Subscription subscription = Observable.just("Hello, World!")
    .subscribe(s -> System.out.println(s));

このリンクである Subscription を以下のように利用することができます.


subscription.unsubscribe();
System.out.println("Unsubscribed=" + subscription.isUnsubscribed());
// "Unsubscribed=true" と出力される

RxJava が unsubscribe を操作することの良いところは, チェインをストップすることができることです. 複雑な Operator のチェインを利用している場合, unsubscribe を使うことでいつでも実行中のコードを停止することができます. 他に処理は必要ありません.

その1で Observable.just() は, onNext() と onComplete() を単純にコールすることに比べて, すこし複雑であることを書きました. その理由は Subscription にあります. それは onNext() をコールする前に, Subscriber がすでに subscribe されているかどうかを実際にチェックするからです.

まとめ

気に止めておいて欲しいのは, これらの話は RxJava の序論です. ここで説明したことよりもっと多くの学ぶべきことがあると思います. たとえば, Backpressure について調べてみてください.

すべてに これらのリアクティブなコードは使う必要はありません. シンプルなロジックにしたい複雑なコードに対して利用しましょう.

これらの説明で このおもしろいフレームワークを始めるには十分だと思います. もっと学びたければ, 公式 RxJava Wiki を読むことをおすすめします. そして「可能性は無限」であることを忘れないでください.

元々は, この記事で最終にする予定でしたが Android 向けの実用的なサンプルのリクエストが多く, その4 を書いています.

[原文] Grokking RxJava, Part 3: Reactive with Benefits

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

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

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

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

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