前提

RxJavaとKotlinを使って今回はお話させて頂きます。

突然ですが問題です

val subject = PublishSubject.create<Int>()
val stream = subject.doOnNext {
    Log.d("yashims85", "value ${it.toString()} on stream")
}
stream.filter({ it > 10 }).subscribe {
    Log.d("yashims85", "${it.toString()} is bigger then 10")
}
stream.filter({ it % 4 == 0 }).subscribe {
    Log.d("yashims85", "${it.toString()} surplus 4 is zero")
}
subject.onNext(12)

上記のコードからは共通のstreamを

メッセージが10より大きければ流れるstream
メッセージが4で割り切れたら流れるstream

の2つに分割したい意図が感じ取れますが、共通部分のvalue 12 on streamは何回出力されるでしょうか？

答え

結果

12-22 14:39:39.654 22766-22766/? D/yashims85: value 12 on stream
12-22 14:39:39.654 22766-22766/? D/yashims85: 12 is bigger then 10
12-22 14:39:39.654 22766-22766/? D/yashims85: value 12 on stream
12-22 14:39:39.654 22766-22766/? D/yashims85: 12 surplus 4 is zero

というわけで2回出力されてしましました。これはobservableのチェーンがsubscribe毎に作られるためです。

しかし、本来は共通部分は1回だけ処理して欲しところですね。。。

そんな時役立つのが、Observable.share()です。

Observable.share

上流のObservableをHot変換し、上流から来たデータを流す別のObservableを下流に提供します。
また、shareで生成されたObservableをObserveする人がいなくなると、上流のObserveも停止します。

つまりは、share()の戻り値のObservableは上流のObservableとは別物なので、streamを共通化出来ます。

試してみよう

share()を追加してみる

val subject = PublishSubject.create<Int>()
val stream = subject.doOnNext {
    Log.d("yashims85", "value ${it.toString()} on stream")
}.share()
stream.filter({ it > 10 }).subscribe {
    Log.d("yashims85", "${it.toString()} is bigger then 10")
}
stream.filter({ it % 4 == 0 }).subscribe {
    Log.d("yashims85", "${it.toString()} surplus 4 is zero")
}
subject.onNext(12)

最初のコードのdoOnNextの後にshare()を追加しただけです。
さて、結果はどうなるでしょうか？

結果

12-22 14:52:04.414 9325-9325/? D/yashims85: value 12 on stream
12-22 14:52:04.414 9325-9325/? D/yashims85: 12 is bigger then 10
12-22 14:52:04.414 9325-9325/? D/yashims85: 12 surplus 4 is zero

はい！共通化できました！

まとめ

streamを分割したい時はshareするとよい。

明日の担当は、採用担当でジンジニアの二つ名を持つjinjineerさんです。わたし気になります！