全てをストリームに感じろ
こんにちは。
最近、RxとかRxJavaとかRxAndroidとかすごい流行ってますよね。
流行に乗って私も勉強し始めたのですが、結構長い間理解できませんでした。（というか、今もできてません）
いろんな記事とか読んでても、「つまりどういうことだってばよ」と意味がわからないことがとても多かったです。特に、
- Observableってなに？Observerパターンと何が違うの？
- ストリームってなに？あのヘンテコな図読めないんだけど？
- 非同期処理が得意なのはわかったけど、それだけ？それしかできないの？
みたいなところがとてもつらかったです。
【翻訳】あなたが求めていたリアクティブプログラミング入門 を読んでても、全てがストリームになる感覚が全然掴めなくて、途中に出てくるこの犬の気持ちがさっぱり分からなくてつらい思いをしたのをよく覚えています。
そんなRx初心者な私が、全てがストリームになる感覚を一瞬感じられた瞬間があったので、そのためにやったことをまとめてみようと思います。
Everything is a stream!!
前提
- RxJava, RxAndroidを使ってAndroidアプリを作ります
- retrolambda使って処理を簡略化してます
- databindingステキ
- Rxの基本的な使い方とか、それぞれの機能の意味などは細かく説明しません
- packageやimport文は省略しています
Rxを使わないでカウントアプリを作ってみる
今回作るのは、すごい簡単なアプリです。
「ボタンを押したら数字が増える」 はいこれだけ。
これを全てストリームに出来なくて、どうしてストリームを理解出来ようか。
最終形はこんな感じ（数字が減るボタンも最終的には用意しましたが、まずは＋ボタンだけ作っていきます）
これをさくっと作ってみましょう。まずはレイアウトから。
<?xml version="1.0" encoding="utf-8"?>
<layout xmlns:android="http://schemas.android.com/apk/res/android"
xmlns:tools="http://schemas.android.com/tools">
<RelativeLayout
android:id="@+id/activity_main"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:paddingBottom="@dimen/activity_vertical_margin"
android:paddingLeft="@dimen/activity_horizontal_margin"
android:paddingRight="@dimen/activity_horizontal_margin"
android:paddingTop="@dimen/activity_vertical_margin"
tools:context="com.litmon.app.rxstudy.MainActivity">
<TextView
android:id="@+id/countTextView"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_alignParentTop="true"
android:layout_centerHorizontal="true"
android:layout_marginTop="152dp"
android:text="0" />
<Button
android:id="@+id/plusButton"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_centerHorizontal="true"
android:layout_centerVertical="true"
android:text="+1" />
</RelativeLayout>
</layout>
じゃあMainActivityの方の処理も書いていきましょう。
数字が増えていけばいいだけなので、解説するほどでもないですが、一応解説すると
- ①
plusButtonが押されたら - ② 数字を１足して
- ③
countTextViewに表示する
という感じです。
public class MainActivity extends AppCompatActivity {
int count = 0;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
ActivityMainBinding binding = DataBindingUtil.setContentView(this, R.layout.activity_main);
// ① plusButtonが押されたら
binding.plusButton.setOnClickListener(v -> {
// ② 数字を1足して
count = count + 1;
// ③ countTextViewに表示する
binding.countTextView.setText(String.valueOf(count));
});
}
}
たったこれだけですね。簡単ですね。
では、これを１つずつストリームにしてストリームを感じていきましょう。
OnClickのイベントをストリームに
まず、OnClickイベントをストリーム化しましょう。
Rxにはいくつかの登場人物がいますが、端折って説明すると以下の3役がいます。
- onsubscribe(発行されるもの。ここが全ての始まり)
- observable(観測されるもの。これがストリームだ)
- subscriber(購読者。最終的に仕事する人)
ストリームを感じるには、ストリームが始まらないと意味がありません。よって、onsubscribeをするものを作成していきます。
onsubscribe
onsubscribeというのは、端的にいうと「イベントの発行元」です。
今回の場合、「OnClick」イベントをストリーム化していくので、「あるViewがClickされたらストリームを進める」onsubscribeを作っていきます。
public class ViewClickOnSubscribe implements Observable.OnSubscribe<View> {
View view;
public ViewClickOnSubscribe(View view) {
this.view = view;
}
@Override
public void call(Subscriber<? super View> subscriber) {
// ①
view.setOnClickListener(v -> {
// ②
subscriber.onNext(v);
});
}
}
RxJavaでは、Observable.OnSubscribeと呼ばれるinterfaceを実装することでonsubscribeを作ることが出来ます。
callメソッドの中で呼び出されている部分が、ストリームのスタート地点で、ストリームが作成されたときに呼び出されます。
上記のコードでは、先ほども説明した通り、
- ① Viewがクリックされたら
- ② ストリームを進める
というコードが書かれています。
ここで subscribe.onNext(v) に渡している引数ですが、これはストリームに流す値を決めるところで、今回は使用しないので何でも良かったのですがとりあえず押されたViewを流しておきました。
（補足）Androidのonsubscribeについて
イベントをonsubscribe化する手順としては上で説明したとおりですが、こんなことを全てのイベントでいちいち作っていくのはとても大変ですよね。
実は、RxBindingというライブラリを使えばそれらを自分で作らなくても良いのですが、今回は勉強のために一から作りました。
https://github.com/JakeWharton/RxBinding
実用するときには、このライブラリを導入してやっていきたいですね。
observable
では、次はMainActivityの方を書き換えていきます。
上で作ったonsubscribeを流すストリーム、observableを作成していきます。
ストリームは流さないと意味がありません。流していきましょう。
public class MainActivity extends AppCompatActivity {
int count = 0;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
ActivityMainBinding binding = DataBindingUtil.setContentView(this, R.layout.activity_main);
//binding.plusButton.setOnClickListener(v -> {
// count = count + 1;
// binding.countTextView.setText(String.valueOf(count));
//});
ViewClickOnSubscribe plusButtonClickOnSubscribe = new ViewClickOnSubscribe(binding.plusButton);
Observable<View> plusButtonClickObservable = Observable.create(plusButtonClickOnSubscribe);
}
}
observableを作成するときには、onsubscribeが必要です。
ストリームを流す大元の存在から、ストリームが生み出されるのだから当然ですね。
今回は、上で作成したViewClickOnSubscribeによって、plusButtonが押されたら値を流すストリームがこれで作成されました。
subscriber
それでは、このストリームから値を受け取って処理をしましょう。
今回の場合は、値を受け取ったら数字を1足してcountTextViewに表示するsubscriberを作成します。
public class MainActivity extends AppCompatActivity {
int count = 0;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
ActivityMainBinding binding = DataBindingUtil.setContentView(this, R.layout.activity_main);
//binding.plusButton.setOnClickListener(v -> {
// count = count + 1;
// binding.countTextView.setText(String.valueOf(count));
//});
ViewClickOnSubscribe plusButtonClickOnSubscribe = new ViewClickOnSubscribe(binding.plusButton);
Observable<View> plusButtonClickObservable = Observable.create(plusButtonClickOnSubscribe);
plusButtonClickObservable.subscribe(v -> {
count = count + 1;
binding.countTextView.setText(String.valueOf(count));
});
}
}
おや、subscriberはどこにいるんでしょうか？
いえいえ、安心してください。subscriberは、lambda式となって私たちの前に現れています。
上のコードの、plusButtonClickObservable.subscribeの中の部分がsubscriberです。
lambda式になっていると少し分かりづらいですね。lambda式になる前のコードを見てみましょう。
plusButtonClickObservable.subscribe(new Action1<View>() {
@Override
public void call(View view) {
count = count + 1;
binding.countTextView.setText(String.valueOf(count));
}
});
こんな感じです。まだ分かりにくい？しょうがないなぁ。。。
plusButtonClickObservable.subscribe(new Subscriber<View>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(View view) {
count = count + 1;
binding.countTextView.setText(String.valueOf(count));
}
});
これならわかりますね。
Subscriberの役割は、ストリームに流れてきた値を受け取って処理をすることです。
「値を受け取る」部分には、今回はViewを流していましたね。なので、onNextの引数の型はViewとなっています。
また、先ほどは説明していませんでしたが、ストリームにはエラーと終了を流すこともできます。
Subscriberはそれらの値を受け取って、上のように処理を分けて書くことができる、というわけです。
今回の例では、エラーを流すことも終了を流すことも必要ないため、特に断りが無い限りは一番上の方法で書いていきます。(一番上の方法は、onNextだけを書くことが出来る)
さて、これで当初の目的である、「OnClickのイベントをストリームに」が達成されました。
この時点で一度実行してみてください。ボタンを押したら、数字が1増えて表示されるという挙動が何一つ変わらず実現されているはずです。
しかし、このままじゃただOnClickListenerをストリームとかいう分かりにくいものに置き換えたに過ぎません。ムダにコードも増えてるし、これじゃやらないほうがマシだ！と言われても仕方ないでしょう。
ということで、Observableの利点を紹介しながらもっとストリームを感じていきましょう。
ストリームをもっと感じよう
先ほどまでのコードでは、ストリームに対してなにも処理を加えませんでした。
ただ流れ続けるストリーム、そんなものに私たちは興味がありません。
ストリームの利点はいくつかありますが、大事なのはやっぱりこれでしょう。
- ストリームを別のストリームに変換することが出来る
- 複数のストリームを1つにまとめることが出来る
1つずつ説明しながらカウントアプリを強化していきます。
ストリームを変換する
これはどういうことかというと、「ストリームに流れてくる値を別の形に書き換えることが出来る」ということです。
Rxには、ストリームに流れてくる値に対して様々な変化を加えることが出来る強力な機能群が揃っています。
たとえば、今回のカウントアプリでは、countという変数を用意していました。
これは、ストリームを変換することで不要となります。
Observable<View> plusButtonClickObservable = Observable.create(plusButtonClickOnSubscribe);
plusButtonClickObservable
.map(v -> 1)
.scan((acc, i) acc + i)
.subscribe(n -> countTextView.setText(String.valueOf(n)));
大分いきなり形が変わって驚いたかと思います。これも1つずつやっていきましょう。
Observable<Integer> plusButtonNumberObservable = plusButtonClickObservable.map(v -> 1)
observable.map を使うと、ストリームに流れてきた値を別の値に変換することが出来ます。これは一番単純で分かりやすいストリームの変換でしょう。
今回は、ストリームに流れてくるViewを1という数字に変換しました。
この変換作業によって、ストリームは新しいplusButtonNumberObservableというストリームに生まれ変わることになります。
Observable<Integer> plusButtonClickCountObservable = plusButtonNumberObservable.scan((acc, i) acc + i);
これもストリームの変換です。observable.scanは、ストリームに流れてきた値を、これまで流れてきた値全てと組み合わせることができます。
これは、非同期や並列処理にも役立つRxJavaの使い方 という記事の画像を見るとすごくよくわかります。
「今までストリームに"流した値"と、今ストリームに"流れてきた値"を組み合わせて、ストリームに値を流す」ことができるということです。
上のコードでは、accにはplusButtonNumberObservableに今まで流した値が、iにはplusButtonClickCountObservableに流れてきた値が入っています。
これを足し合わせると、次にaccに流れる値がplusButtonClickCountObservableストリームに流れる、という形です。
分かりづらいかもしれないので、表にしてまとめてみました。
| plusButtonClickCountObservableに流れてきた値 | plusButtonNumberObservableに流した値 | plusButtonNumberObservableに流れる値 |
|---|---|---|
| 1 | - | 1 |
| 1 | 1 | 2 |
| 1 | 2 | 3 |
| 1 | 3 | 4 |
| 1 | 4 | 5 |
| 1 | 5 | 6 |
これを見れば分かる通り、plusButtonNumberObservableのストリームにはボタンを押した数が入っていくことが見て取れますね。
最終的に、これをsubscribeして数字を表示することで数字をカウントすることが出来ます。
plusButtonClickCountObservable.subscribe(n -> countTextView.setText(String.valueOf(n)));
このように、ストリームを別のストリームに変換することが出来るのが大きな特徴です。
カウントアプリの例じゃ分かりづらかったかもしれませんが、たとえばObserverパターンやPub/Subパターンなどでは、こういった「値の変換」は自由に行うことが出来ず、受け取る側でそれぞれ変換する必要があります。
また、変換途中のストリームが独立して分けられるので、複数のsubscriberに対して途中のストリームを流すということもできます。
こうした柔軟な対応が行えるのは、Rxの1つの大きな利点だと思います。
また、通信処理などを書いていると、レスポンスで受け取った値を使ってさらにリクエストを送りたい、といったことをしたいときが多くあると思います。（俗にいう、callback地獄というやつです）
ストリームの変換を行えば、受け取った値を別の通信ストリームに変換するなんて造作も無いことです。
callback地獄が一点、ストリーム天国へと昇華します。
クールですね。
複数のストリームを1つにまとめることが出来る
Rxのもう1つの大きな特徴として、複数のストリームを1つにまとめることができます。
これが個人的には最大の利点だと思っていて、ストリームを感じるための大きなきっかけとなったものでした。
とりあえず、カウントアプリにマイナスボタンを付けながら、複数のストリームをまとめてストリームを感じましょう。
レイアウトをさくっと修正して
<?xml version="1.0" encoding="utf-8"?>
<layout xmlns:android="http://schemas.android.com/apk/res/android"
xmlns:tools="http://schemas.android.com/tools">
<RelativeLayout
android:id="@+id/activity_main"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:paddingBottom="@dimen/activity_vertical_margin"
android:paddingLeft="@dimen/activity_horizontal_margin"
android:paddingRight="@dimen/activity_horizontal_margin"
android:paddingTop="@dimen/activity_vertical_margin"
tools:context="com.litmon.app.rxstudy.MainActivity">
<TextView
android:id="@+id/countTextView"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_alignParentTop="true"
android:layout_centerHorizontal="true"
android:layout_marginTop="152dp"
android:text="0" />
<Button
android:id="@+id/plusButton"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_centerHorizontal="true"
android:layout_centerVertical="true"
android:text="+1" />
<Button
android:id="@+id/minusButton"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:text="-1"
android:layout_below="@+id/plusButton"
android:layout_alignLeft="@+id/plusButton"
android:layout_alignStart="@+id/plusButton" />
</RelativeLayout>
</layout>
MainActivityをさくっと修正しましょう。
public class MainActivity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
ActivityMainBinding binding = DataBindingUtil.setContentView(this, R.layout.activity_main);
Observable<View> plusButtonClickObservable = Observable.create(new ViewClickOnSubscribe(binding.plusButton));
Observable<Integer> plusNumberObservable = plusButtonClickObservable
.map(v -> 1)
.scan((acc, i) -> acc + i);
Observable<View> minusButtonClickObservable = Observable.create(new ViewClickOnSubscribe(binding.minusButton));
Observable<Integer> minusNumberObservable = minusButtonClickObservable
.map(v -> -1)
.scan((acc, i) -> acc + i);
// ここ！ここが重要！
Observable<Integer> countObservable = Observable.combineLatest(plusNumberObservable, minusNumberObservable,
(plus, minus) -> plus + minus);
countObservable.subscribe(n -> {
binding.countTextView.setText(String.valueOf(n));
});
}
}
plusボタンの方は上で説明したとおりですね。
minusボタンの方も、plusボタンと同じようにストリームを作成しましょう。
ここで重要な部分は、以下の部分です。
Observable<Integer> countObservable Observable.combineLatest(plusNumberObservable, minusNumberObservable,
(plus, minus) -> plus + minus);
ここで、ストリームの合成を行っています。
plusNumberObservableとminusNumberObservable、2つのストリームをかけ合わせて、新しいストリームを作ることが出来ます。
今回は、Observable.combineLatestという機能で、2つのストリームのうちどちらかに値が流れてきたときに、両方のストリームに流れていた最新の値を取り出して、新しい値を流すストリームを作成することが出来ます。
これも分かりづらいと思ったので、表にしました。
| 行った操作 | plusNumberObservableに流れた値 | minusNumberObservableに流れた値 | countObservableに流れる値 |
|---|---|---|---|
| plusボタンクリック | 1 | - | 1 + 0 = 1 |
| plusボタンクリック | 2 | - | 2 + 0 = 2 |
| minusボタンクリック | - | -1 | 2 + (-1) = 1 |
| plusボタンクリック | 3 | - | 3 + (-1) = 2 |
| minusボタンクリック | - | -2 | 3 + (-2) = 1 |
いかがでしょうか？
plusボタンをクリックしたときにはminusNumberObservableには値は流れていませんが、combineLatestで作られたストリームにはplusNumberObservableとminusNumberObservableで流れた最新の値を足し合わせた値が流れます。
あとは、このストリームでの値をcountTextViewに表示すれば、＋ボタンとーボタンの実装が完成します。
countObservable.subscribe(n -> {
binding.countTextView.setText(String.valueOf(n));
});
この、「複数のストリームをかけ合わせて１つのストリームにする」のは、使いようによってはとても大きな効力を発揮します。
たとえば、「AというチェックボックスとBというフィールドに正しい値が入っていれば、Cというボタンが押せるようになる」という処理を書きたいときに、普通ならそれぞれのListenerに同じような処理をそれぞれ書く必要がありますが、ストリームにしてしまえばそれも簡単に実現できます。
こういうやつですね。
「チェックAがONならば、項目Bは入力必須とする」という Validation を RxJava + RxAndroid でやる
また、非同期処理をしているときに、「Aの非同期処理とBの非同期処理がどっちも終わった後にCの処理をしたい」という要望は昔からとても多くあると思いますが、これもストリームにしてしまえば簡単に実現することが出来ます。
とてもクールですね。
いかがでしたか？
カウントアプリを使って、ストリームを感じてみました。
カウントアプリという単純な構造のアプリにRxを適用することは、ストリームを感じるのにとても有用だったと個人的に思っています。
この犬のように悟りを開くまでにはまだたくさんの障壁がありますが、だいぶハードルが下がったと思います。
これまで一切ストリームを感じれなかった人も、この記事を読んで少しでもストリームを感じることができたら幸いです。
Everything is a stream!!
以上です。
参考：Rxを勉強するときに読んだ記事たち
全てをストリームに感じるまでに読んだ記事とか。とても分かりやすい記事ばかりでいつも助かっています。