Rxjavaのliftについて

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](http://reactivex.io/RxJava/javadoc/rx/Observable.html#lift(rx.Observable.Operator))

```
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に変換しています。

```java
.map(text -> {
   return Integer.valueOf(text);
})
```

ではmapの実装を覗いてみましょう。`map`とは 「Tを受け取ってRを返す`Func1`を引数として受け取り、`Observable<R>`を返す」メソッドのようです。

```java
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`の実装を覗いていみましょう。
 
 ```java
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`の実装を覗いてみます。

```java
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 `の実装で重要なのは以下の３行です。

```java
Subscriber<? super T> st = hook.onLift(operator).call(o);
st.onStart();
onSubscribe.call(st);
```

今回は`hook.onLift(operator)`の部分の説明は省略します。これは、`operator.call(o)`と等価であると考えて下さい。`Subscriber#onStart`も実質何もしないのと考えて良いので、より簡単にするとこうなります。

```java
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は自分で理解できても他人に説明するのがものすごく難しいと感じているのですが、もしこのエントリーがどなたかの理解の助けになれたら、幸いです。