RxJava AdventCalendarの初日は、liftについて、紹介します。
初学者なので間違った理解があるかもしれませんが、その際はコメントなどでご指摘頂けると幸いです。
また、一部説明の端折る部分があると思いますが、簡略化のためなのでご了承下さい
liftとはなにか
Rxjavaはメソッドチェーンで処理を繋げて書くのが一般的ですが、このメソッドチェーンを実現しているのがliftです。
普段の開発でliftを見かけることは少ないと思いますが、mapやfilterなど殆どのオペレーターの中身はliftとOperator<R, T> の組み合わせで作られています。liftを理解すると、自分でカスタムオペレーターも作ることも出来ます。 [Implementing Your Own Operators](https://github.com/
ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators)
liftのjavadocによる説明は以下のようです。javadoc
Lifts a function to the current Observable and returns a new Observable that when subscribed to will pass the values of the current Observable through the Operator function.
In other words, this allows chaining Observers together on an Observable for acting on the values within the Observable.
observable.map(...).filter(...).take(5).lift(new OperatorA()).lift(new OperatorB(...)).subscribe()
If the operator you are creating is designed to act on the individual items emitted by a source Observable, use lift. If your operator is designed to transform the source Observable as a whole (for instance, by applying a particular set of existing RxJava operators to it) use Observable.compose(rx.Observable.Transformer<? super T, ? extends R>).
(javadocの中で登場するcomposeについては、12/5のエントリーで紹介したいと思います。)
liftの仕組み
mapを例にとってliftが何をしているのか確認しましょう。
mapとは値を受け取り、加工して返すオペレーターです。以下の処理では、渡ってきた文字列をintに変換しています。
.map(text -> {
return Integer.valueOf(text);
})
ではmapの実装を覗いてみましょう。mapとは 「Tを受け取ってRを返すFunc1を引数として受け取り、Observable<R>を返す」メソッドのようです。
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
return lift(new OperatorMap<T, R>(func));
}
mapの実装で登場する OperatorMapとは先ほど出てきた、Operator<R, T>のサブクラスです。ではOperatorMapの実装を覗いていみましょう。
public final class OperatorMap<T, R> implements Observable.Operator<R, T> {
private final Func1<? super T, ? extends R> transformer;
public OperatorMap(Func1<? super T, ? extends R> transformer) {
this.transformer = transformer;
}
@Override
public Subscriber<? super T> call(final Subscriber<? super R> o) {
return new Subscriber<T>(o) {
@Override
public void onCompleted() {
o.onCompleted();
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onNext(T t) {
try {
o.onNext(transformer.call(t));
} catch (Throwable e) {
Exceptions.throwOrReport(e, this, t);
}
}
};
}
}
思ったよりシンプルですね。Subscriber<? super R>を引数として受け取り、Subscriber<? super T>を返すcallメソッドがあるだけです。
OperatorMapの場合は、「mapが引数として受け取ったFunc」をonNextの引数に適応し、「OperatorMap#callの引数のo(Subscriber)#onNext」にその値を渡しています。つまりOperatorとは「値を受け取って->ゴニョゴニョして->また発信する」役割があるようです。 OperatorMap以外のOperatorも処理の複雑さは様々ですが似たようなことを行っています。
ここまでで、Operatorとは「Subscriberを引数として受け取り、Subscriberを返すFunc1」であることがわかりました。
では、OperatorMap#callで渡されるSubscriberはどこから来るのでしょう。そこで、liftの実装を覗いてみます。
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return new Observable<R>(new OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = hook.onLift(operator).call(o);
try {
st.onStart();
onSubscribe.call(st);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
});
}
liftの中では、新しいObservableを作って返してしています。先ほどのSubscriberは新しく作られたObservableの引数のOnSubscribeから渡ってきていたんですね。lift の実装で重要なのは以下の３行です。
Subscriber<? super T> st = hook.onLift(operator).call(o);
st.onStart();
onSubscribe.call(st);
今回はhook.onLift(operator)の部分の説明は省略します。これは、operator.call(o)と等価であると考えて下さい。Subscriber#onStartも実質何もしないのと考えて良いので、より簡単にするとこうなります。
Subscriber<? super T> st = operator.call(o);
onSubscribe.call(st);
oというのはmapよりメソッドチェーンの後ろで接続されれるSubscriberだと考えてると、構造が大体わかってきましたね。
oが渡されると、oをもとに「operatorが適応されたSubscriber」がOperator<R, T>#callによって作られます。そして「operatorが適応されたSubscriber」を使い、親のObservable(lift内部でnewしたものではない方)の処理をcallします。つまり末端から始まり(コード上では下から上)に向けて、OnSubscribe#callが伝播するようになっています。このような仕組みでメソッドチェーンが実現されているんですね。
まとめ
Rxjavaを学ぶと一番初めに躓くのがliftだと思います。liftが理解できると、今まで魔法のように思っていたsubscribeOnやobserveOnもliftとOpertor<R,T>の組わせで実現されているので(ものすごく難しいですが)読み解くことが出来るようになります。Rxjavaは自分で理解できても他人に説明するのがものすごく難しいと感じているのですが、もしこのエントリーがどなたかの理解の助けになれたら、幸いです。