Rxjavaソース解析

こちらの文章は主にRxjavaを使って、mapの方法とflatmapの方法のソースコードを理解しています。RxJavaの真髄だと思います。RxJavaの詳細な使用については、ネットワークに多くの資料があります。ここでは[ ]の文章を紹介します。
http://gank.io/post/560e15be2dca930e00da1083
次に本題に入ります。まずmapの方法を見ます。
map方法解析：

public final  Observable map(Func1 super T, ? extends R> func) {    
    return lift(new OperatorMap(func));
}

OperatomapはOperatorインターフェースの実現類です。Operatorインターフェースを見てください。

public interface Operator extends Func1, Subscriber super T>> {    
    // cover for generics insanity
}

OperatorはまたFnc 1を継承しています。このインターフェースは一つの方法R call(T t);だけがOperatorMapのcall方法を見て実現します。

public Subscriber super T> call(final Subscriber super R> o) {
        return new Subscriber(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);
                }
            }

        };
    }

そしてlift関数を見ます

public final  Observable lift(final Operator extends R, ? super T> operator) {
        return new Observable(new OnSubscribe() {
            @Override
            public void call(Subscriber super R> o) {
                try {
                    Subscriber super T> st = hook.onLift(operator).call(o);
                    try {
                        // new Subscriber created and being subscribed with so 'onStart' it
                        st.onStart();
                        onSubscribe.call(st);
                    } catch (Throwable e) {
                        // localized capture of errors rather than it skipping all operators 
                        // and ending up in the try/catch of the subscribe method which then
                        // prevents onErrorResumeNext and other similar approaches to error handling
                        Exceptions.throwIfFatal(e);
                        st.onError(e);
                    }
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    // if the lift function failed all we can do is pass the error to the final Subscriber
                    // as we don't have the operator available to us
                    o.onError(e);
                }
            }
        });
    }

簡単に言えば、三つのことをしました。1.新しいObservableを作成して、元Observableからのイベントを引き継ぎます。2.hook.onLift(operator).call(o)はOperatorMapのcall方法を実行して、新しいSubscriberに戻ります。3.onSubscribe.call(st)新しいSubscriberは元のObservableに伝えます。元Observable送信イベントでは、新しいSubscriberを呼び出すオンネクスト方法であり、transformer.call(t)すなわちmap(Fnc 1)メソッドでパラメータFnc 1のコール方法を実行し、元のSubscriberのオンネクスト方法を実行します。
faltmapメソッド解析
相変わらずです。まずfaltmapのソースを貼ります。

public final  Observable flatMap(Func1 super T, ? extends Observable extends R>> func) {
        if (getClass() == ScalarSynchronousObservable.class) {
            return ((ScalarSynchronousObservable)this).scalarFlatMap(func);
        }
        return merge(map(func));
}

ここでmerge(map(func))を実行します。ここは先ほどのmap方法です。はい、map方法は言いません。前の説明を見てもいいです。ここにヒントを与えてください。 map Observableを見てください。

public final static  Observable merge(Observable extends Observable extends T>> source) {
        if (source.getClass() == ScalarSynchronousObservable.class) {
            return ((ScalarSynchronousObservable)source).scalarFlatMap((Func1)UtilityFunctions.identity());
        }
        return source.lift(OperatorMerge.instance(false));
}

ここではmerge まで実行されます。実は以前のmapメソッドで紹介されたlift方法です。ここのsource.lift(OperatorMerge.instance(false))は、新しいObservableがすべてのオリジナルObservableから送信されたイベントを受信し、新しいObservableを構成すると理解できる。その後liftの方法を実行して、以前のmap方法の分析から、ここでOperatorの中のcall方法を実行して、元のSubscriberの中のcall方法を実行します。ここのOperatorはOperatoMergeです。この種類の中のコール方法を見てください。

public Subscriber> call(final Subscriber super T> child) {
        MergeSubscriber subscriber = new MergeSubscriber(child, delayErrors, maxConcurrent);
        MergeProducer producer = new MergeProducer(subscriber);
        subscriber.producer = producer;
        
        child.add(subscriber);
        child.setProducer(producer);
        
        return subscriber;
}

ここでいくつかの操作を経て、最終的にはMergeProducerのrequest方法に進みます。

public void request(long n) {
    if (n > 0) {
        if (get() == Long.MAX_VALUE) {
            return;
        }
        BackpressureUtils.getAndAddRequest(this, n);
        subscriber.emit();
    } else 
    if (n < 0) {
        throw new IllegalArgumentException("n >= 0 required");
    }
}

ここで、このsource Observableはすべてのイベントを送信することに注意する。このように解釈します。1.すべてのオリジナルObservableのイベントを受信し、新たなObservable 2を構成します。新しいObservableはすべてのイベントを送信します。3.元Subscriberは新しいイベントを受信してから処理します。
まとめ:
mapであろうとflatmapであろうと、実はすべて転換の思想を運用したのです。

元イベントの配布プロセスを遮断する。

は中間処理動作を増加させる（mapはコールバック方式を追加し、flatmapはイベント収集再送信を追加する）。

元のイベント配信フローに戻ってイベントを処理する。