オブザーバブル

[Exposed=*]
interface Subscriber {
  undefined next(any value);
  undefined error(any error);
  undefined complete();
  undefined addTeardown(VoidFunction teardown);

  // True after the Subscriber is created, up until either
  // complete()/error() are invoked, or the subscriber unsubscribes. Inside
  // complete()/error(), this attribute is true.
  readonly attribute boolean active;

  readonly attribute AbortSignal signal;
};

next(value) メソッドの手順は以下の通りです：

1. this の active が false なら、return。

2. this の 関連グローバルオブジェクト が Window オブジェクトであり、 その 関連付けられた Document が 完全にアクティブ でない場合、return。

3. internal observers copy を this の 内部オブザーバのコピーとする。

   注: 内部オブザーバリストのコピーを作成して反復処理することで、 もし1つの 内部オブザーバの next steps が サブスクリプションを引き起こした場合でも、イテレーション中にオブザーバリストが変化しないようにしています。

4. internal observers copy の各 observer について：

   1. observer の next steps を value で実行する。

      アサート：例外は投げられなかった。

      注: ここで例外が投げられないのは、内部オブザーバの next steps が スクリプト提供のコールバックのラッパーの場合、process observer 手順がこれらのコールバックを呼び出す際に例外をキャッチし、グローバルに報告するロジックでラップしているためです。

      next steps が仕様アルゴリズムの場合、 これらの手順はこのアサートを満たすため、例外を外部に投げません。

error(error) メソッドの手順は以下の通りです：

1. this の active が false なら、例外を報告し、error と this の 関連グローバルオブジェクト で、return。

2. this の 関連グローバルオブジェクト が Window オブジェクトであり、 その 関連付けられた Document が 完全にアクティブでない場合、return。

3. クローズ this。

4. internal observers copy を this の 内部オブザーバ のコピーとする。

5. internal observers copy の各 observer について：

   1. observer の error steps を error で実行する。

      アサート：例外は投げられなかった。

      注: これが正しい理由については、next() の説明を参照してください。

complete() メソッドの手順は以下の通りです：

1. this の active が false なら、return。

2. this の 関連グローバルオブジェクト が Window オブジェクトであり、 その 関連付けられた Document が 完全にアクティブ でない場合、return。

3. クローズ this。

4. internal observers copy を this の 内部オブザーバ のコピーとする。

5. internal observers copy の各 observer について：

   1. observer の complete steps を実行する。

      アサート：例外は投げられなかった。

      注: これが正しい理由については、next() の説明を参照してください。

addTeardown(teardown) メソッドの手順は以下の通りです：

1. this の 関連グローバルオブジェクト が Window オブジェクトであり、 その 関連付けられた Document が 完全にアクティブ でない場合、return。

2. this の active が true なら、append で teardown を this の ティアダウンコールバックリストに追加する。

3. それ以外の場合は、invoke teardown を «» と "report" で呼び出す。

サブスクリプションをクローズするために、 Subscriber subscriber と、オプションの any reason を受け取った場合、以下の手順を実行します：

1. subscriber の active が false なら、return。

   これは再入呼び出しを防ぐためのガードです。これは「プロデューサー主導の」購読解除の場合に発生することがあります。以下の例を参照：

   const outerController = new AbortController();
   const observable = new Observable(subscriber => {
     subscriber.addTeardown(() => {
       // 2.) This teardown executes inside the "Close" algorithm, while it’s
       //     running. Aborting the downstream signal run its abort algorithms,
       //     one of which is the currently-running "Close" algorithm.
       outerController.abort();
     });
   
     // 1.) This immediately invokes the "Close" algorithm, which
     //     sets subscriber.active to false.
     subscriber.complete();
   });
   
   observable.subscribe({}, {signal: outerController.signal});
   

2. subscriber の active ブール値を false に設定する。

3. abort シグナル を subscriber の サブスクリプションコントローラー に対して reason が与えられていればそれを使って実行する。

4. subscriber の ティアダウンコールバック を 逆順で反復して各 teardown について：

   1. もし subscriber の 関連グローバルオブジェクト が Window オブジェクトであり、その 関連付けられた Document が 完全にアクティブ でない場合は、この手順を中止する。

      注: この手順は繰り返し実行されます。なぜなら各 teardown が上記の Document を非アクティブにする可能性があるためです。

   2. teardown を «» と "report" で呼び出す。

// SubscribeCallbackは、Observableの「作成者」のコードが存在する場所です。
// subscribe()が呼び出されたときに呼び出され、新しい購読をセットアップします。
callback SubscribeCallback = undefined (Subscriber subscriber);
callback ObservableSubscriptionCallback = undefined (any value);

dictionary SubscriptionObserver {
  ObservableSubscriptionCallback next;
  ObservableSubscriptionCallback error;
  VoidFunction complete;
};

callback ObservableInspectorAbortHandler = undefined (any value);

dictionary ObservableInspector {
  ObservableSubscriptionCallback next;
  ObservableSubscriptionCallback error;
  VoidFunction complete;

  VoidFunction subscribe;
  ObservableInspectorAbortHandler abort;
};

typedef (ObservableSubscriptionCallback or SubscriptionObserver) ObserverUnion;
typedef (ObservableSubscriptionCallback or ObservableInspector) ObservableInspectorUnion;

dictionary SubscribeOptions {
  AbortSignal signal;
};

callback Predicate = boolean (any value, unsigned long long index);
callback Reducer = any (any accumulator, any currentValue, unsigned long long index);
callback Mapper = any (any value, unsigned long long index);
// Mapperとの違いは戻り値の型のみで、このコールバックはシーケンス内の各要素を訪問するためだけに使われ、変換はしません。
callback Visitor = undefined (any value, unsigned long long index);

// このコールバックは`any`を返し、`Observable`への変換セマンティクスを通じて`Observable`に変換される必要があります。
callback CatchCallback = any (any value);

[Exposed=*]
interface Observable {
  constructor(SubscribeCallback callback);
  undefined subscribe(optional ObserverUnion observer = {}, optional SubscribeOptions options = {});

  // valueが以下のいずれかの場合、ネイティブObservableを構築します:
  //   - Observable
  //   - AsyncIterable
  //   - Iterable
  //   - Promise
  static Observable from(any value);

  // Observableを返すオペレーター。仕様書の「Operators」セクション参照。
  //
  // takeUntil()はpromise, iterable, async iterable, 他のobservableを消費できます。
  Observable takeUntil(any value);
  Observable map(Mapper mapper);
  Observable filter(Predicate predicate);
  Observable take(unsigned long long amount);
  Observable drop(unsigned long long amount);
  Observable flatMap(Mapper mapper);
  Observable switchMap(Mapper mapper);
  Observable inspect(optional ObservableInspectorUnion inspectorUnion = {});
  Observable catch(CatchCallback callback);
  Observable finally(VoidFunction callback);

  // Promiseを返すオペレーター。
  Promise<sequence<any>> toArray(optional SubscribeOptions options = {});
  Promise<undefined> forEach(Visitor callback, optional SubscribeOptions options = {});
  Promise<boolean> every(Predicate predicate, optional SubscribeOptions options = {});
  Promise<any> first(optional SubscribeOptions options = {});
  Promise<any> last(optional SubscribeOptions options = {});
  Promise<any> find(Predicate predicate, optional SubscribeOptions options = {});
  Promise<boolean> some(Predicate predicate, optional SubscribeOptions options = {});
  Promise<any> reduce(Reducer reducer, optional any initialValue, optional SubscribeOptions options = {});
};

new Observable(callback) コンストラクターの手順は以下の通りです：

1. this の subscribe コールバック を callback に設定する。

   注: このコールバックは、後でsubscribe() が呼ばれた時に呼び出されます。

subscribe(observer, options) メソッドの手順は：

1. Observableを購読する：this に対して observer と options を使う。

デフォルトのエラーアルゴリズムは、 any error を受け取り、以下の手順を実行します：

1. 例外を報告：error と 現在のrealmの グローバルオブジェクト で。

注: このデフォルトを別途定義することで、 本仕様のすべてのネイティブな Observable購読（spec proseから購読、subscribe()メソッドを経由しない場合）で 冗長な手順定義が不要になります。

Observableに変換するために、 any value を受け取った場合、以下の手順を実行します：

注: このアルゴリズムはWeb IDLの from() メソッドから分離しています。仕様プローズからWeb IDLバインディングを経由せずに値を変換できるようにするためです。

1. Type(value)が Objectでない場合は、TypeErrorを TypeError で投げる。

   注: これはプリミティブ型がIterableに変換されること（例：String）を防ぎます。 WICG/observable#125の議論参照。

2. Observableからの変換: value の specific typeが Observable の場合、value を返す。

3. Async Iterableからの変換: asyncIteratorMethod を ? GetMethod(value, %Symbol.asyncIterator%) とする。

   注: GetMethodを使うのは、async iteratorプロトコル対応のみを調べたいからで、 実装されていない時にthrowしたくないためです。 GetIteratorは(a)プロトコル未実装、(b)実装済だがcallableでない・getterがthrowする、 両方でthrowします。GetMethodなら後者だけでthrowします。

4. asyncIteratorMethod が undefined または null なら、Iterableからの変換へジャンプ。

5. nextAlgorithm を、Subscriber subscriber と Iterator Record iteratorRecord を受け取る以下の手順とする：

   1. subscriber の subscription controller の signal が abortedならreturn。

   2. nextPromise を Promiseまたはundefinedで初期値undefinedとする。

   3. nextCompletion を IteratorNext(iteratorRecord)とする。

      注: IteratorNextを使うのは、IteratorStepValueが iteratorのnext()が即座に値を返すobjectを期待するためですが、 async iteratorの場合はnext()がPromise/thenableを返す想定なので、Promiseでラップして値取得に反応します。

   4. nextCompletionがthrow completionの場合：

      1. Assert: iteratorRecordの[[Done]]はtrue。

      2. nextPromiseに拒否されたpromiseで nextRecordの[[Value]]を設定。

   5. それ以外でnextRecordがnormal completionの場合は nextPromiseに解決されたpromiseで nextRecordの[[Value]]を設定。

      注: nextRecordの[[Value]]自体が すでにPromise でない場合に行います。

   6. Promiseがsettleしたら反応：

      • nextPromiseが値iteratorResultでfulfillされたら：

        1. Type(iteratorResult) がObjectでない場合は、subscriberの error() メソッドをTypeError で呼び出し、この手順を中止。

        2. doneをIteratorComplete(iteratorResult)とする。

        3. doneがthrow completionなら subscriberの error() メソッドをdoneの[[Value]]で呼び出し、この手順を中止。

        4. doneの[[Value]]がtrueなら、subscriberの complete() を呼び出し、この手順を中止。

        5. valueをIteratorValue(iteratorResult)とする。

        6. valueがthrow completionなら subscriberの error() メソッドをvalueの[[Value]]で呼び出し、この手順を中止。

        7. subscriberのnext() をvalueの[[Value]]で呼び出す。

        8. nextAlgorithmをsubscriberとiteratorRecordで再実行。

      • nextPromiseが理由rでrejectされたら、subscriberの error() をrで呼び出す。

6. 新しい Observable を返す。subscribe コールバックは Subscriber subscriberを受け取り、以下の処理をするアルゴリズムとする：

   1. subscriberのsubscription controllerの signalが abortedならreturn。

   2. iteratorRecordCompletionをGetIterator(value, async)とする。

      注: %Symbol.asyncIterator% のgetterが再度呼ばれる場合がありますが、これは極端なケースでありテスト期待値に合わせます。 詳細はissue#127の議論参照。 プロトコル自体を呼び出してIterator Recordを取得します。

   3. iteratorRecordCompletionがthrow completionなら subscriberの error() をiteratorRecordCompletionの[[Value]]で呼び出し、手順中止。

      注: この場合だけ、購読に同期してerror() を呼びます。他ケースではPromiseでrejectしてmicrotaskで非同期に報告します。 この同期エラー伝播は言語構造―たとえばfor-await-ofループと同様の挙動です。

   4. iteratorRecordを! iteratorRecordCompletionとする。

   5. Assert: iteratorRecordは Iterator Recordである。

   6. subscriberのsubscription controllerの signalが abortedならreturn。

   7. 以下のabortアルゴリズムを追加： subscriberの subscription controllerの signalに追加：

      1. AsyncIteratorClose(iteratorRecord, NormalCompletion(subscriberの subscription controllerのabort reason))

   8. nextAlgorithmをsubscriberとiteratorRecordで実行。

7. Iterableからの変換: iteratorMethodを? GetMethod(value, %Symbol.iterator%) とする。

8. iteratorMethodがundefinedなら、Promiseからの変換へジャンプ。

   そうでなければ、新しい Observable を返す。subscribe コールバックは Subscriber subscriberを受け取り、以下の手順を行うアルゴリズムとする：

   1. subscriberのsubscription controllerの signalが abortedならreturn。

   2. iteratorRecordCompletionをGetIterator(value, sync)とする。

   3. iteratorRecordCompletionがthrow completionなら subscriberの error() をiteratorRecordCompletionの[[Value]]で呼び出し、手順中止。

   4. iteratorRecordを! iteratorRecordCompletionとする。

   5. subscriberのsubscription controllerの signalが abortedならreturn。

   6. 以下のabortアルゴリズムを追加： subscriberの subscription controllerの signalに追加：

      1. IteratorClose(iteratorRecord, NormalCompletion(UNUSED))

   7. true である間:

      1. next を IteratorStepValue(iteratorRecord) とする。

      2. もし next が throw completion なら、 subscriber の error() メソッドを next の [[Value]] で実行し、break する。

      3. next を ! で next に設定する。

      4. もし next が done なら:

         1. アサート: iteratorRecord の [[Done]] は true である。

         2. subscriber の complete() を実行する。

         3. Return。

      5. subscriber の next() を next で実行する。

      6. もし subscriber の subscription controller の signal が aborted なら break する。

9. Promiseから: IsPromise(value) が true なら、

   1. 新しい Observable を返す。その subscribe callback は、Subscriber subscriber を受け取り、以下を行うアルゴリズムである：

      1. value に反応する：

         1. もし value が v で fulfilled された場合：

            1. subscriber の next() メソッドを v で実行する。

            2. subscriber の complete() メソッドを実行する。

         2. もし value が r で rejected された場合は、 subscriber の error() メソッドを r で実行する。

10. TypeError を throw する。

テスト

• observable-from.any.js （ライブテスト） （ソース）

Observable に subscribe する とき、ObserverUnion または internal observer observer と、SubscribeOptions options を受け取り、以下の手順を実行する：

注: このアルゴリズムは Web IDL の subscribe() メソッドから分離されており、 仕様のプローズが subscribe を Web IDL バインディングを通さずに呼び出せるようにしている。 JavaScript によってプロパティが変更されうるオブジェクトでは、「内部」プローズは Web IDL バインディングを経由してはいけない。 § 2.3.3 Promiseを返すオペレーター でこの使用例を参照。

1. this の 関連グローバルオブジェクト が Window オブジェクトで、 その 関連付けられた Document が 完全にアクティブ でない場合は return。

2. internal observer を新しい internal observer とする。

3. observer を以下のように処理する：

   1. もし observer が ObservableSubscriptionCallback なら

      internal observer の next steps を、any value を受け取る以下の手順に設定：

      1. observer を «value» と "report" で呼び出す。

      もし observer が SubscriptionObserver なら

      1. もし observer の next が 存在すれば、 internal observer の next steps を、any value を受け取る以下の手順に設定：

         1. observer の next を «value» と "report" で呼び出す。

      2. もし observer の error が 存在すれば、 internal observer の error steps を、any error を受け取る以下の手順に設定：

         1. observer の error を «error» と "report" で呼び出す。

      3. もし observer の complete が 存在すれば、 internal observer の complete steps を以下の手順に設定：

         1. observer の complete を «» と "report" で呼び出す。

      もし observer が internal observer なら

      internal observer を observer に設定する。

4. アサート: internal observer の error steps は デフォルトの error アルゴリズム か、 error コールバック関数を呼び出すアルゴリズムである。

5. this の weak subscriber が null でなく、かつ this の weak subscriber の active が true なら：

   1. subscriber を this の weak subscriber とする。

   2. internal observer を subscriber の internal observers に追加する。

   3. もし options の signal が 存在すれば：

      1. もし options の signal が aborted なら、 internal observer を subscriber の internal observers から削除する。

      2. それ以外の場合は、以下の abort アルゴリズム を options の signal に追加する：

         1. もし subscriber の active が false なら、この手順を中止する。

         2. internal observer を subscriber の internal observers から削除する。

         3. もし subscriber の internal observers が 空なら、 subscriber を close する。 options の signal の abort reason を使う。

   4. Return。

6. subscriber を 新しい Subscriber とする。

7. internal observer を subscriber の internal observers に追加する。

8. this の weak subscriber を subscriber に設定する。

9. もし options の signal が 存在すれば：

   1. もし options の signal が aborted なら、 subscriber を close する options の signal の abort reason を使う。

   2. それ以外の場合は 以下の abort アルゴリズム を options の signal に追加する：

      1. もし subscriber の active が false なら、この手順を中止する。

      2. internal observer を subscriber の internal observers から削除する。

      3. もし subscriber の internal observers が 空なら、 subscriber を close する。 options の signal の abort reason を使う。

10. もし this の subscribe callback が SubscribeCallback なら、 subscriber を «subscriber» と "rethrow" で呼び出す。

    もし 例外 E が投げられた場合は、 subscriber の error() を E で呼び出す。

11. それ以外の場合は this の subscribe callback の手順を subscriber で実行する。

from(value) メソッドの手順は以下の通りです：

1. value を Observable に変換した結果を返す。例外があれば再スローする。

takeUntil(value) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. notifier を value を Observable に変換した結果とする。

3. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   このメソッドは2つの Observable にsubscribeすることを伴います。 (1) notifier、(2) sourceObservable。次の状況で両方の購読を解除します：

   1. notifier が値（"next" または "error"）の発行を開始した場合、この場合は notifier から購読解除します。 必要な情報を得たので、これ以上値を発行し続ける必要がありません。同時に sourceObservable からも購読解除します。 なぜなら、observable への購読を手動で終了するため、sourceObservable の値を observable に送る必要がなくなるためです。

   2. sourceObservable が error() または complete() を自ら実行した場合、この場合は notifier から購読解除します。 なぜなら observable が sourceObservable の値をミラーする必要がなくなったためです。 sourceObservable の購読解除は不要です、なぜなら自ら完了したためです。

   1. notifierObserver を新しい internal observer とし、次のように初期化する：

      next steps

      subscriber の complete() メソッドを実行する。

      注: これにより sourceObservable の購読も解除されます。 これは sourceObservable が "外側" subscriber の subscription controller の signal で購読されているためです。 この signal は上記または下記で subscriber の complete() が呼ばれると abort されます。

      error steps

      subscriber の complete() メソッドを実行する。

      注: notifier Observable が自ら complete した場合は subscriber を complete する必要はありません。 この場合 observable は sourceObservable を継続してミラーし続けます。

   2. options を新しい SubscribeOptions とし、signal は subscriber の subscription controller の signal とする。

   3. notifier に notifierObserver と options でsubscribeする。

   4. subscriber の active が false なら、return。

      注: notifier が同期的に値を発行した場合、sourceObservable の subscribe callback は一度も呼ばれません。 ただし notifier が同期的に complete のみした場合は、subscriber の active は true のままなので、sourceObservable への購読が行われ、observable は途切れることなくミラーします。

   5. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      渡された value を使って subscriber の next() メソッドを実行する。

      error steps

      渡された error を使って subscriber の error() メソッドを実行する。

      complete steps

      subscriber の complete() メソッドを実行する。

      注: sourceObserver は基本的にパススルーであり、sourceObservable の発行するすべてをミラーします。 ただし sourceObservable が notifier より先に完了した場合は notifier から購読を解除します。

   6. sourceObservable に sourceObserver と options でsubscribeする。

4. observable を返す。

テスト

• observable-takeUntil.any.js （ライブテスト） （ソース）
• observable-takeUntil.window.js （ライブテスト） （ソース）

map(mapper) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. idx を unsigned long long、初期値0とする。

   2. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. mapper を «渡された value, idx» と "rethrow" で呼び出し、返り値を mappedValue とする。

         もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

      2. idx をインクリメントする。

      3. subscriber の next() メソッドを mappedValue で実行する。

      error steps

      渡された error で subscriber の error() メソッドを実行する。

      complete steps

      subscriber の complete() メソッドを実行する。

   3. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   4. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

テスト

• observable-map.any.js （ライブテスト） （ソース）
• observable-map.window.js （ライブテスト） （ソース）

filter(predicate) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. idx を unsigned long long、初期値0とする。

   2. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. predicate を «渡された value, idx» と "rethrow" で呼び出し、返り値を matches とする。

         もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

      2. idx を idx + 1 に設定する。

      3. もし matches が true なら、subscriber の next() メソッドを value で実行する。

      error steps

      渡された error で subscriber の error() メソッドを実行する。

      complete steps

      subscriber の complete() メソッドを実行する。

   3. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   4. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

テスト

• observable-filter.any.js （ライブテスト） （ソース）

take(amount) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. remaining を amount とする。

   2. もし remaining が 0 なら、subscriber の complete() メソッドを実行し、この手順を中止する。

   3. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. 渡された value で subscriber の next() メソッドを実行する。

      2. remaining をデクリメントする。

      3. もし remaining が 0 なら、subscriber の complete() メソッドを実行する。

      error steps

      渡された error で subscriber の error() メソッドを実行する。

      complete steps

      subscriber の complete() メソッドを実行する。

   4. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   5. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

テスト

• observable-take.any.js （ライブテスト） （ソース）

drop(amount) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. remaining を amount とする。

   2. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. もし remaining が > 0 なら、remaining をデクリメントし、この手順を中止する。

      2. アサート: remaining は 0 である。

      3. 渡された value で subscriber の next() メソッドを実行する。

      error steps

      渡された error で subscriber の error() メソッドを実行する。

      complete steps

      subscriber の complete() メソッドを実行する。

   3. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   4. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

テスト

• observable-drop.any.js （ライブテスト） （ソース）

flatMap(mapper) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. idx を unsigned long long、初期値0とする。

   2. outerSubscriptionHasCompleted を boolean、初期値 false とする。

   3. queue を新しい list（any型値）、初期値は空とする。

      注: queue は observable が直前に subscribe された Observable がまだ完了していない間に sourceObservable から発行される Observable を保存するために使われます。

   4. activeInnerSubscription を boolean、初期値 false とする。

   5. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. もし activeInnerSubscription が true なら：

         1. queue に value を追加する。

            注: この value は、現在 subscribe 中の Observable が 完了した後に処理されます。

      2. それ以外の場合：

         1. activeInnerSubscription を true に設定する。

         2. flatmap process next value steps を value, subscriber, mapper、そして 参照として queue, activeInnerSubscription, outerSubscriptionHasCompleted, idx で実行する。

            注: この flatmap process next value steps は value から導出された Observable に subscribe し、 その購読が非アクティブになるまで値を処理し続けます（エラーまたは完了）。"inner" の Observable が完了した場合、 処理手順は queue の次の値で再帰的に自分自身を呼び出します。

            値が 存在しない 場合、処理手順は終了し、 activeInnerSubscription の設定を解除し、 sourceObservable から発行される将来の値が正しく処理されるようにします。

      error steps

      渡された error で subscriber の error() メソッドを実行する。

      complete steps

      1. outerSubscriptionHasCompleted を true にする。

         注: activeInnerSubscription が true の場合、下記の手順では subscriber を complete しません。 その場合は flatmap process next value steps が、 "inner" の購読が非アクティブになった後、queue が 空であれば subscriber を complete する責任を持ちます。

      2. activeInnerSubscription が false かつ queue が 空なら、subscriber の complete() メソッドを実行する。

   6. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   7. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

flatmap process next value steps は、any value、Subscriber subscriber、Mapper mapper、および次の参照を受け取る：list型 any 値の queue、boolean の activeInnerSubscription、boolean の outerSubscriptionHasCompleted、unsigned long long の idx の場合、次の手順を実行する：

1. mappedResult を mapper を «value, idx» と "rethrow" で呼び出した結果とする。

   もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

2. idx を idx + 1 に設定する。

3. innerObservable を from() を mappedResult で呼び出した結果とする。

   もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

   from() を直接呼び出すべきではありません。例外を適切に処理する内部アルゴリズムを呼び出し、subscriber にパイプしたいです。

4. innerObserver を新しい internal observer とし、次のように初期化する：

   next steps

   渡された value で subscriber の next() メソッドを実行する。

   error steps

   渡された error で subscriber の error() メソッドを実行する。

   complete steps

   1. queue が空でなければ：

      1. nextValue を queue の先頭要素とし、remove で queue から削除する。

      2. flatmap process next value steps を nextValue、subscriber、mapper、 参照として queue、activeInnerSubscription で実行する。

   2. それ以外の場合：

      1. activeInnerSubscription を false に設定する。

         注: activeInnerSubscription は参照型なので、 以降 "outer" の Observable（sourceObservable）から発行される値が正しく処理されます。

      2. outerSubscriptionHasCompleted が true なら、 subscriber の complete() メソッドを実行する。

         注: "outer" Observable が完了しているが、 まだ subscriber を complete していません。これは、まだ完了していない "inner" の Observable（つまり innerObservable）が キューに存在していたためです。今まさに完了したということです。

5. innerOptions を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

6. innerObservable に innerObserver と innerOptions で subscribe する。

switchMap(mapper) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. idx を unsigned long long、初期値0とする。

   2. outerSubscriptionHasCompleted を boolean、初期値 false とする。

   3. activeInnerAbortController を AbortController または null、初期値 null とする。

      注: AbortController はこのアルゴリズムの next steps のみで新しい値が代入され、 null になるのは switchmap process next value steps で "inner" Observable が完了またはエラーになった時です。現在 "inner" の購読があるかどうかの判定に使います。"complete steps" も参照します。

   4. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. activeInnerAbortController が null でなければ、 signal abort する。

         注: これにより前回の値から導出された "inner" Observable から購読解除される。 すぐに新しい値から導出された "inner" Observable に購読する。

      2. activeInnerAbortController を 新しい AbortController に設定する。

      3. switchmap process next value steps を value、subscriber、mapper、参照として activeInnerAbortController、outerSubscriptionHasCompleted、idx で実行する。

         注: switchmap process next value steps は、 value から導出される Observable に subscribe し、 その購読が非アクティブになる（エラー/完了）か、activeInnerAbortController が abort されるまで値を処理します。

      error steps

      渡された error で subscriber の error() メソッドを実行する。

      complete steps

      1. outerSubscriptionHasCompleted を true に設定する。

         注: activeInnerAbortController が null でない場合、直ちに subscriber を complete しない。 その場合は switchmap process next value steps が inner の購読が完了した時点で subscriber を complete する。

      2. activeInnerAbortController が null なら、subscriber の complete() メソッドを実行する。

   5. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   6. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

switchmap process next value steps は、any value、Subscriber subscriber、Mapper mapper、および次の参照を受け取る：AbortController activeInnerAbortController、boolean の outerSubscriptionHasCompleted、 unsigned long long の idx の場合、次の手順を実行する：

1. mappedResult を mapper を «value, idx» と "rethrow" で呼び出した結果とする。

   もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

2. idx を idx + 1 に設定する。

3. innerObservable を from() を mappedResult で呼び出した結果とする。

   もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

4. innerObserver を新しい internal observer とし、次のように初期化する：

   next steps

   渡された value で subscriber の next() メソッドを実行する。

   error steps

   渡された error で subscriber の error() メソッドを実行する。

   注: この時点で activeInnerAbortController を null にする必要はありません。 なぜなら subscriber の error() を呼び出すことで "outer" の Observable から購読解除されるため、以降値が push されることはありません。

   complete steps

   1. outerSubscriptionHasCompleted が true なら、 subscriber の complete() メソッドを実行する。

   2. それ以外の場合は activeInnerAbortController を null に設定する。

      注: この変数は参照型なので、 switchMap complete steps に "inner" の購読がないことを知らせます。

5. innerOptions を新しい SubscribeOptions とし、 signal は、create a dependent abort signal を «activeInnerAbortController の signal、subscriber の subscription controller の signal»、 AbortSignal、 current realm で生成したものとする。

6. innerObservable に innerObserver と innerOptions で subscribe する。

inspect(inspectorUnion) メソッドの手順は以下の通りです：

1. subscribe callback を VoidFunction もしくは null（初期値 null）とする。

2. next callback を ObservableSubscriptionCallback もしくは null（初期値 null）とする。

3. error callback を ObservableSubscriptionCallback もしくは null（初期値 null）とする。

4. complete callback を VoidFunction もしくは null（初期値 null）とする。

5. abort callback を ObservableInspectorAbortHandler もしくは null（初期値 null）とする。

6. inspectorUnion を次のように処理する：

   もし inspectorUnion が ObservableSubscriptionCallback なら

   1. next callback を inspectorUnion に設定する。

   もし inspectorUnion が ObservableInspector なら

   1. subscribe が 存在すれば、subscribe callback にそれを設定。

   2. next が 存在すれば、next callback にそれを設定。

   3. error が 存在すれば、error callback にそれを設定。

   4. complete が 存在すれば、complete callback にそれを設定。

   5. abort が 存在すれば、abort callback にそれを設定。

7. sourceObservable を this とする。

8. observable を 新しい Observable とし、 その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. subscribe callback が null でなければ、invoke で «» と "rethrow" で呼び出す。

      もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

      注: この結果、sourceObservable は決して subscribe されません。

   2. abort callback が null でなければ、subscriber の subscription controller の signal に次の abort algorithm を追加する：

      1. abort callback を «subscriber の subscription controller の signal の abort reason» と "report" で呼び出す。

   3. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      1. next callback が null でなければ、 next callback を «渡された value» と "rethrow" で呼び出す。

         もし 例外 E が投げられた場合：

         1. abort callback を subscriber の subscription controller の signal から削除する。

            注: abort callback は consumer-initiated の解除だけに使うため、プロデューサーによるエラーや完了時は呼ばれないようにする必要があります。

            Chromiumの実装と一致していますが、もともと渡された SubscribeOptions の signal への参照を保持し、その abort 時だけ abort callback を呼び出すべきかも。要調査。

         2. subscriber の error() メソッドを E で実行し、この手順を中止する。

      2. 渡された value で subscriber の next() メソッドを実行する。

      error steps

      1. abort callback を subscriber の subscription controller の signal から削除する。

      2. error callback が null でなければ、 error callback を «渡された error» と "rethrow" で呼び出す。

         もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

      3. 渡された error で subscriber の error() メソッドを実行する。

      complete steps

      1. abort callback を subscriber の subscription controller の signal から削除する。

      2. complete callback が null でなければ、 complete callback を «» と "rethrow" で呼び出す。

         もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

      3. subscriber の complete() メソッドを実行する。

   4. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   5. sourceObservable に sourceObserver と options で subscribe する。

9. observable を返す。

テスト

• observable-inspect.any.js （ライブテスト） （ソース）

catch(callback) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、 その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      渡された value で subscriber の next() メソッドを実行する。

      error steps

      1. callback を «渡された error» と "rethrow" で呼び出し、 返り値を result とする。

         もし 例外 E が投げられた場合は、 subscriber の error() を E で実行し、この手順を中止する。

      2. innerObservable を from() を result で呼び出した結果とする。

         もし 例外 E が投げられた場合は、 subscriber の error() メソッドを E で実行し、この手順を中止する。

         from() を直接呼び出すべきではありません。例外を適切に処理する内部アルゴリズムを呼び出し、subscriber にパイプしたいです。

      3. innerObserver を新しい internal observer とし、次のように初期化する：

         next steps

         渡された value で subscriber の next() メソッドを実行する。

         error steps

         渡された error で subscriber の error() メソッドを実行する。

         complete steps

         subscriber の complete() メソッドを実行する。

      4. innerOptions を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

      5. innerObservable に innerObserver と innerOptions で subscribe する。

         注: innerObservableへのsubscribeは sourceObservableから購読解除せず安全に実施可能です。なぜなら error steps 内で呼ばれるので sourceObservable の購読は既に完了しています。

      complete steps

      subscriber の complete() メソッドを実行する。

   2. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   3. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

finally(callback) メソッドの手順は以下の通りです：

1. sourceObservable を this とする。

2. observable を 新しい Observable とし、 その subscribe callback は Subscriber subscriber を受け取るアルゴリズムである：

   1. subscriber の addTeardown() メソッドを callback で実行する。

   2. sourceObserver を新しい internal observer とし、次のように初期化する：

      next steps

      渡された value で subscriber の next() メソッドを実行する。

      error steps

      1. 渡された error で subscriber の error() メソッドを実行する。

      complete steps

      1. subscriber の complete() メソッドを実行する。

   3. options を新しい SubscribeOptions とし、 signal は subscriber の subscription controller の signal とする。

   4. sourceObservable に sourceObserver と options で subscribe する。

3. observable を返す。

toArray(options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. options の signal が null でない場合：

   1. options の signal が aborted なら：

      1. p を options の signal の abort reason で reject する。

      2. p を返す。

   2. abort algorithm を options の signal に追加：

      1. p を options の signal の abort reason で reject する。

      注: ここでは p を reject するだけです。this Observable の購読も自動的にcloseされます。"inner" Subscriberが closeされるためです。

3. values を新しい listとする。

4. observer を新しい internal observer とし、次のように初期化：

   next steps

   渡された value を values に追加する。

   error steps

   p を渡された error で reject する。

   complete steps

   p を values で resolve する。

5. this に observer と options で subscribe する。

6. p を返す。

テスト

• observable-toArray.any.js （ライブテスト） （ソース）

forEach(callback, options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. visitor callback controller を新しい AbortControllerとする。

3. internal options を新しい SubscribeOptions とし、 signal は dependent abort signal（visitor callback controller の signal、options の signal（nullでなければ））で生成する。 AbortSignal、 current realm を使う。

   多くの trivial な internal observer は値をチェーンに流すだけで購読を制御しません。 このオペレーターでは下記 observer の next steps が callback が例外を投げた場合、基礎となる購読を abort する責任を持ちます。なので "Observableへのsubscribe" に渡す SubscribeOptions の signal は options の signal と、かつ AbortSignal を持つ AbortController の dependent signal にする必要がある。

4. internal options の signal が aborted なら：

   1. p を internal options の signal の abort reason で reject する。

   2. p を返す。

5. abort algorithm を internal options の signal に追加：

   1. p を internal options の signal の abort reason で reject する。

      注: p の reject は internal options の signal に結び付けられています。 options の signal の abort時の microtaskが先にキューされ、p の reject handler より先に実行されます。

6. idx を unsigned long long（初期値0）とする。

7. observer を新しい internal observer とし、次のように初期化：

   next steps

   1. callback を «渡された value, idx» と "rethrow" で呼び出す。

      もし 例外 E が投げられた場合は、 p を E で reject し、visitor callback controller を E で abort する。

   2. idx をインクリメントする。

   error steps

   p を渡された error で reject する。

   complete steps

   p を undefined で resolve する。

8. this に observer と internal options で subscribe する。

9. p を返す。

テスト

• observable-forEach.any.js （ライブテスト） （ソース）

every(predicate, options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. controller を新しい AbortControllerとする。

3. internal options を新しい SubscribeOptions とし、 signal は dependent abort signal（controller の signal、options の signal（nullでなければ））で生成する。 AbortSignal、 current realm を使う。

4. internal options の signal が aborted なら：

   1. p を internal options の signal の abort reason で reject する。

   2. p を返す。

5. abort algorithm を internal options の signal に追加：

   1. p を internal options の signal の abort reason で reject する。

6. idx を unsigned long long（初期値0）とする。

7. observer を新しい internal observer とし、次のように初期化：

   next steps

   1. predicate を «渡された value, idx» と "rethrow" で呼び出し、返り値を passed とする。

      もし 例外 E が投げられた場合は、 p を E で reject し、controller を E で abort する。

   2. idx を idx + 1 に設定する。

   3. passed が false なら p を false で resolve し、controller を abort する。

   error steps

   p を渡された error で reject する。

   complete steps

   p を true で resolve する。

8. this に observer と internal options で subscribe する。

9. p を返す。

first(options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. controller を新しい AbortControllerとする。

3. internal options を新しい SubscribeOptions とし、 signal は dependent abort signal（controller の signal、options の signal（nullでなければ））で生成する。 AbortSignal、 current realm を使う。

4. internal options の signal が aborted なら：

   1. p を internal options の signal の abort reason で reject する。

   2. p を返す。

5. abort algorithm を internal options の signal に追加：

   1. p を internal options の signal の abort reason で reject する。

6. internal observer を新しい internal observer とし、次のように初期化：

   next steps

   1. p を渡された value で resolve する。

   2. controller を abort する。

   error steps

   Reject p を渡された error で拒否する。

   complete steps

   p を新しい RangeError で reject する。

   注: これは、ソース Observable が 1つも値を発行する前に complete した場合のみ到達します。

7. this に internal observer と internal options で subscribe する。

8. p を返す。

last(options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. options の signal が null でない場合：

   1. options の signal が aborted なら：

      1. p を options の signal の abort reason で reject する。

      2. p を返す。

   2. abort algorithm を options の signal に追加：

      1. p を options の signal の abort reason で reject する。

3. lastValue を any または null（初期値 null）とする。

4. hasLastValue を boolean（初期値 false）とする。

5. observer を新しい internal observer とし、次のように初期化：

   next steps

   1. hasLastValue を true に設定する。

   2. lastValue を渡された value に設定する。

   error steps

   p を渡された error で reject する。

   complete steps

   1. hasLastValue が true なら p を lastValue で resolve する。

      1. それ以外なら p を新しい RangeError で reject する。

         注: first() の注を参照。

6. this に observer と options で subscribe する。

7. p を返す。

find(predicate, options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. controller を新しい AbortControllerとする。

3. internal options を新しい SubscribeOptions とし、 signal は dependent abort signal（controller の signal、options の signal（nullでなければ））で生成する。 AbortSignal、 current realm を使う。

4. internal options の signal が aborted なら：

   1. p を internal options の signal の abort reason で reject する。

   2. p を返す。

5. abort algorithm を internal options の signal に追加：

   1. p を internal options の signal の abort reason で reject する。

6. idx を unsigned long long（初期値0）とする。

7. observer を新しい internal observer とし、次のように初期化：

   next steps

   1. predicate を «渡された value, idx» と "rethrow" で呼び出し、返り値を passed とする。

      もし 例外 E が投げられた場合は、 p を E で reject し、controller を E で abort する。

   2. idx を idx + 1 に設定する。

   3. passed が true なら p を value で resolve し、controller を abort する。

   error steps

   p を渡された error で reject する。

   complete steps

   p を undefined で resolve する。

8. this に observer と internal options で subscribe する。

9. p を返す。

some(predicate, options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. controller を新しい AbortControllerとする。

3. internal options を新しい SubscribeOptions とし、 signal は dependent abort signal（controller の signal、options の signal（nullでなければ））で生成する。 AbortSignal、 current realm を使う。

4. internal options の signal が aborted なら：

   1. p を internal options の signal の abort reason で reject する。

   2. p を返す。

5. abort algorithm を internal options の signal に追加：

   1. p を internal options の signal の abort reason で reject する。

6. idx を unsigned long long（初期値0）とする。

7. observer を新しい internal observer とし、次のように初期化：

   next steps

   1. predicate を «渡された value, idx» と "rethrow" で呼び出し、返り値を passed とする。

      もし 例外 E が投げられた場合は、 p を E で reject し、controller を E で abort する。

   2. idx を idx + 1 に設定する。

   3. passed が true なら p を true で resolve し、controller を abort する。

   error steps

   p を渡された error で reject する。

   complete steps

   p を false で resolve する。

8. this に observer と internal options で subscribe する。

9. p を返す。

reduce(reducer, initialValue, options) メソッドの手順は以下の通りです：

1. p を新しい promiseとする。

2. controller を新しい AbortControllerとする。

3. internal options を新しい SubscribeOptions とし、 signal は dependent abort signal（controller の signal、options の signal（nullでなければ））で生成する。 AbortSignal、 current realm を使う。

4. internal options の signal が aborted なら：

   1. p を internal options の signal の abort reason で reject する。

   2. p を返す。

5. abort algorithm を internal options の signal に追加：

   1. p を internal options の signal の abort reason で reject する。

6. idx を unsigned long long（初期値0）とする。

7. accumulator を initialValue（与えられていれば）で初期化し、なければ未初期化とする。

8. observer を新しい internal observer とし、次のように初期化：

   next steps

   1. accumulator が未初期化（つまり initialValue が渡されていない）なら、 accumulator を渡された value に設定し、idx を idx + 1 にし、この手順を中止する。

      注: この場合 reducer は this が最初に出す currentValue では呼ばれません。 2つ目以降の値が発行されたとき、reducer をその値（currentValue）と 最初の値（accumulator）で呼び出します。

   2. reducer を «accumulator（accumulator）, 渡された value（currentValue）, idx（index）»と "rethrow" で呼び出し、返り値を result とする。

      もし 例外 E が投げられた場合は、 p を E で reject し、controller を E で abort する。

   3. idx を idx + 1 に設定する。

   4. accumulator を result に設定する。

   error steps

   p を渡された error で reject する。

   complete steps

   1. accumulator が "unset" 以外なら p を accumulator で resolve する。

      それ以外なら p を TypeError で reject する。

9. this に observer と internal options で subscribe する。

10. p を返す。

dictionary ObservableEventListenerOptions {
  boolean capture = false;
  boolean passive;
};

partial interface EventTarget {
  Observable when(DOMString type, optional ObservableEventListenerOptions options = {});
};

when(type, options) メソッドの手順は以下の通りです：

1. this の 関連グローバルオブジェクト が Window オブジェクトで、その 関連付けられた Document が 完全にアクティブ でない場合は return。

2. event target を this とする。

3. observable を 新しい Observable とし、次のように初期化する：

   subscribe callback

   Subscriber subscriber を受け取るアルゴリズムで、次の手順を実行する：

   1. event target が null なら、この手順を中止する。

      注: event target は購読時点で ガベージコレクションされている可能性を考慮しています。

   2. subscriber の subscription controller の signal が aborted なら、この手順を中止する。

   3. event listener を追加する。対象は event target、 event listener は次の通り：

      type

      type

      callback

      新しい Web IDL EventListener インスタンスを生成し、1引数の Event 型 event を受け取る関数への参照とする。この関数は observable event listener invoke algorithm を subscriber と event で実行する。

      capture

      options の capture

      passive

      options の passive が 存在すればその値、なければ null。

      once

      false

      signal

      subscriber の subscription controller の signal

      注: event listener は subscription controller の signal が aborted になった時 必ずクリーンアップされます（エンジンの所有モデルによらず）。

4. observable を返す。

observable event listener invoke algorithm は Subscriber subscriber と Event event を受け取り、次の手順を実行する：

1. subscriber の next() メソッドを event で実行する。