Observable の仕組み

Observable というと、値を見張り続ける watcher ルーチンや、変化をポーリングするループ、通知のタイミングを決めるスケジューラのような「生きているもの」を想像しがちです。SynState の Observable はそのどれでもありません。裏で動いているものは何もありません。 subscribe はリストに関数を追加し、next はそのリストの関数を同期的に呼び出してから return します。Observable は「呼び出し元と呼び出し先の紐付け」の抽象化であり、その紐付けは Observable のグラフを組み立てた時点で静的に決まります。

最小の Observable

次の Observable は subscribe と next だけを持ちます。細部は SynState の実装と異なりますが、仕組みは同じです：

type Subscription = Readonly<{ unsubscribe: () => void }>;

type MiniObservable<A> = Readonly<{
    subscribe: (onNext: (value: A) => void) => Subscription;
    next: (value: A) => void;
}>;

const createObservable = <A,>(): MiniObservable<A> => {
    // The whole "reactive machinery": a list of functions.
    const mut_subscribers = new Set<(value: A) => void>();

    return {
        subscribe: (onNext) => {
            mut_subscribers.add(onNext); // Remember the function. That's all.

            return {
                unsubscribe: () => {
                    mut_subscribers.delete(onNext);
                },
            };
        },

        next: (value) => {
            // Call every registered function, right here, synchronously.
            for (const onNext of mut_subscribers) {
                onNext(value);
            }
        },
    };
};

const count = createObservable<number>();

// A derived value is one more registered function that calls `next`.
const doubled = createObservable<number>();

count.subscribe((n) => {
    doubled.next(n * 2);
});

const mut_log: string[] = [];

const subscription = doubled.subscribe((n) => {
    mut_log.push(`doubled: ${n}`);
});

mut_log.push('before next');

count.next(1);

mut_log.push('after next');

// The subscriber ran inside `count.next(1)`, before it returned.
assert.deepStrictEqual(mut_log, ['before next', 'doubled: 2', 'after next']);

subscription.unsubscribe();

count.next(2);

// Removed from the list, so never called again.
assert.deepStrictEqual(mut_log, ['before next', 'doubled: 2', 'after next']);

subscribe は何も起動しません。関数を保存し、それを取り除く手段を返すだけです。next は保存された関数をループで呼ぶ、ごく普通の関数呼び出しです。間にキューもタイマーもマイクロタスクも挟まらないので、ログを見ると subscriber は 'before next' と 'after next' の間、つまり count.next(1) の内側で実行されています。unsubscribe で取り除かれた関数は、単にリストにもう存在しません。

doubled は、派生値も同じ仕組みの繰り返しであることを示しています。count に登録された「doubled.next を呼ぶ関数」にすぎません。count.next(1) はその関数を呼び、それが doubled.next(2) を呼び、それがログを取る関数を呼びます。subscribe を呼んだ時点で敷かれた、ただの関数呼び出しの同期的な連鎖です。

SynState も同じように振る舞う

同じグラフを SynState で書くと、同じログになります：

import { map, source } from 'synstate';

const count = source<number>();

const doubled = count.pipe(map((n) => n * 2));

const mut_log: string[] = [];

const subscription = doubled.subscribe((n) => {
    mut_log.push(`doubled: ${n}`);
});

mut_log.push('before next');

count.next(1);

mut_log.push('after next');

assert.deepStrictEqual(mut_log, ['before next', 'doubled: 2', 'after next']);

subscription.unsubscribe();

SynState が最小版に加えているものも、この構図を変えません：

• Observable は現在値も保持しており（getSnapshot()）、新しい subscriber にはそれがすぐに渡されます。
• map や combine のような派生 Observable は subscribe で連鎖しているわけではありません。生成時に親へ自分を登録し、最も深い親より 1 つ大きい深さを持ちます。ルートの next が始めるような更新の中では、発行した Observable がそれぞれ自分の子を深さごとのバケットからなるキューに入れ、その更新を始めたノードがバケットを浅いほうから順に空にしていきます。そのため各子はすべての親が済んだあとに一度だけ実行され、これによって SynState はグリッチを回避しています（SynState がグリッチを解決した方法）。深さはグラフの構築時に決まり、キューを空にするのもやはり同期的なループです。
• 完了処理（complete）も扱います。

中核は ObservableBaseClass です。subscribe は subscriber の Set に追加し、setNext は値を保存して各 subscriber を for ループで呼び出します。

非同期性はどこから来るのか

伝搬が同期的なら、timer、counter、fromPromise、debounce はどうやって動いているのでしょうか。これらは next を呼ぶ側です。時間に基づくソースは、setTimeout / setInterval のコールバックから更新が開始される Observable です。fromPromise は Promise のコールバックから開始します。debounce や switchMap のような非同期演算子は、そうしたコールバック（switchMap の場合は内側の Observable の subscriber）から自分自身の伝搬を開始します（switchMap が購読した時点で内側の Observable がすでに持っている値、たとえば just(x) の値は、新しい伝搬ではありません。内側の Observable を作った更新の中でそのまま流すので、下流がその値をソースの古い値と並べて見ることはありません。switchMap を作った時点でソースがすでに持っている値はそのときに射影し、まだ何も switchMap を観測していないので、内側の Observable が出す値は保持するだけです）。タイマーは Observable ではなく JavaScript ランタイムが持っており、コールバックが更新を開始した時点から先は、上で見たとおりすべて同期的に実行されます。

ここから次のことが直接導かれます：

• 何もしていないグラフのコストはゼロです。 next が呼ばれなければ、どのコードも動きません。
• 新しい状態はすぐに読めます。 next が return した後は、グラフ内のどの Observable の getSnapshot() も更新後の値を返します。
• next 1 回が 1 回の完結した伝搬です。 SynState は複数の next 呼び出しを 1 回の更新にまとめません。
• subscriber は呼び出し元のスタック上で実行されます。 遅い subscriber は next の return を遅らせ、subscriber が投げた例外は next を呼んだコードに伝わります。