RxJavaの簡単なコード(Observable.just("Hello").subscribe())の内部コードを読んでみるメモ

何かRxJavaの内部コードに詳しくなる足がかりになればと思い読んでみました
RxJavaのコードを読むのは初挑戦なので、何か見当違いなことを言っていたら教えて下さい。

以下のコードの中を見ていきます。
動かすと"Hello"と出力されます。

```java
        Observable.just("Hello").subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                
            }

            @Override
            public void onNext(String s) {
                Log.d("MainActivity", s);
            }
        });
```

Javaの実行の順番的にまずjust()の実行が最初です。

## just()

```java
Observable.just("Hello")
```


```java
    public static <T> Observable<T> just(final T value) {
        return ScalarSynchronousObservable.create(value);
    }
```

ScalarSynchronousObservableクラスをインスタンス化しました

```java
    public static <T> ScalarSynchronousObservable<T> create(T t) {
        return new ScalarSynchronousObservable<T>(t);
    }
```

ScalarSynchronousObservableのコンストラクタはこちら
ちょっと難しくなってきましたね、、
OnSubscribeクラスの無名クラスをインスタンス化しています。
callをオーバーライドして渡しています。
callではSubscriberを受け取り、createProducer()した結果をsetProducer()しているようです。
いったんcall文の中が呼ばれるまで、一旦callメソッドの中の実装は置いておきましょう、、

```java
    protected ScalarSynchronousObservable(final T t) {
        super(new OnSubscribe<T>() {

            @Override
            public void call(Subscriber<? super T> s) {
                s.setProducer(createProducer(s, t));
            }

        });
        this.t = t;
    }
```

親クラスのコンストラクタはこんな感じで、普通にOnSubscribeクラスをメンバ変数に突っ込むだけみたいです。

```java
    protected Observable(OnSubscribe<T> f) {
        this.onSubscribe = f;
    }
```

ここまでで行われたこと
**OnSubscribeクラスのインスタンスを持ったScalarSynchronousObservableクラスのインスタンスが作られた。**


続きです。

## subscribe()

ScalarSynchronousObservable#subscribe()を呼び出します。
ちなみにこのときのthisはさきほどのScalarSynchronousObservableです。


```java
    public final Subscription subscribe(Subscriber<? super T> subscriber) {
        return Observable.subscribe(subscriber, this);
    }
```

Observable.subscribe()は長めなので最初から少しずつ読んでいきます。渡したsubscriberのonStart()を呼び出します。onStart()は継承していないかぎり空の実装です。

```java
    private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
     // 例外処理のため省略
        
        // new Subscriber so onStart it
        subscriber.onStart();
        
```


次に、SafeSubscriberでラップします(なぜかは力不足で分かっていません)

```java
        /*
         * See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
         * to user code from within an Observer"
         */
        // if not already wrapped
        if (!(subscriber instanceof SafeSubscriber)) {
            // assign to `observer` so we return the protected version
            subscriber = new SafeSubscriber<T>(subscriber);
        }
```

hookというメンバ変数のonSubscribeStartを呼び出した後callを呼び出します。


```java
        // The code below is exactly the same an unsafeSubscribe but not used because it would 
        // add a significant depth to already huge call stacks.
        try {
            // allow the hook to intercept and/or decorate
            hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
```

hookはこんな感じでインスタンス化されていて、うまく他のオブジェクトなどが差し込めるようになっているようでした

```java
    static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
```

ちなみに今回の実体はRxJavaObservableExecutionHookDefaultというクラスのインスタンスになっていました。
![image](https://qiita-image-store.s3.amazonaws.com/0/27388/a3007105-4e44-6a10-7d0d-05db1bbc9d57.png)

この実装は以下のようになっており、渡されたOnSubscribeクラスのインスタンスのonSubscribeをそのまま返します。

```java
    public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {
        // pass-thru by default
        return onSubscribe;
    }
```

そのため、observable.onSubscribeのcall(subscriber)メソッドを呼び出します。

```java
            // allow the hook to intercept and/or decorate
            hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
```

observable.onSubscribeの実体はjust()でやったとおり、以下のようになっていました。

```java
    protected ScalarSynchronousObservable(final T t) {
        super(new OnSubscribe<T>() {

            @Override
            public void call(Subscriber<? super T> s) {
                s.setProducer(createProducer(s, t));
            }

        });
        this.t = t;
    }
```

つまりこの` s.setProducer(createProducer(s, t));`が実行されます。


createProducer()はこのようになっています。


```java
    /**
     * Creates a scalar producer depending on the state of STRONG_MODE.
     * @param <T> the type of the scalar value
     * @param s the target subscriber
     * @param v the value to emit
     * @return the created Producer
     */
    static <T> Producer createProducer(Subscriber<? super T> s, T v) {
        if (STRONG_MODE) {
            return new SingleProducer<T>(s, v);
        }
        return new WeakSingleProducer<T>(s, v);
    }
    
```

WeakSingleProducerを単純に作るだけっぽいです。(デバッガでWeakSingleProducerになることを確認)

```java
    static final class WeakSingleProducer<T> implements Producer {
        final Subscriber<? super T> actual;
        final T value;
        boolean once;
        
        public WeakSingleProducer(Subscriber<? super T> actual, T value) {
            this.actual = actual;
            this.value = value;
        }
```

ちょっとここで何がなんのオブジェクトを持っているのか整理したいです。
Observable(実体はScalarSynchronousObservable)がOnSubscribeのインスタンスのonSubscribeをメンバ変数に持っていて、
WeakSingleProducerは自分でインスタンス化したSubscriber(変数名actual)とjustで渡された文字列"hello"(変数名value)をもっています

`Subscriber#setProducer()`はどうなっているかというと以下のようになっていてsubscriberにsetProducerします。(ちょっとSafeSubscriber関連の処理を省略)


```java
    public void setProducer(Producer p) {
        // いろいろ省略
        producer.request(Long.MAX_VALUE);
        // いろいろ省略
    }
```

WeakSingleProducer.request(long n)が呼びだされてやっとSubscriberのonNext()が呼び出されました！
**つまりSubscriberにsetProducer()してあげると、SubscriberがonNext()やonComplete()を呼び出すProducer.request()メソッドを呼び出してくれるようです(ちょっと分かりにくい)**

```java
        @Override
        public void request(long n) {
            if (n != 0L) {
                once = true;
                Subscriber<? super T> a = actual;
                if (a.isUnsubscribed()) {
                    return;
                }
                T v = value;
                try {
                    a.onNext(v);
                } catch (Throwable e) {
                    Exceptions.throwOrReport(e, a, v);
                    return;
                }
                
                if (a.isUnsubscribed()) {
                    return;
                }
                a.onCompleted();
            }
```

## まとめ
ちょっとRxJavaを触ったことがある人は、OnSubscribeクラスのサブクラスを作ったことがあるかもしれません。
基本的には普通にその仕組を使うのですが、Producerを使っている点が異なりました。
それを使ってうまくデータを伝えるクラス(汎用性がある？)がProducerインターフェースを実装したクラスなのかな？という感じでした。
ちゃんとコードの呼び出しで追えて気持ちが良かったので、良かったです。
もう一段階mapなどを挟んだ場合もちゃんと追いたいです(すごく簡単な例でもちょっと複雑でした。。)