Observable

1. If this 's active is false, then return.

2. If this 's relevant global object is a Window object, and its associated Document is not fully active , then return.

3. If this 's next algorithm is not null, then run this 's next algorithm given value .

   Assert : No exception was thrown .

1. If this 's active is false, report the exception error , then return.

2. If this 's relevant global object is a Window object, and its associated Document is not fully active , then return.

3. Let error algorithm be this 's error algorithm .

4. Close this .

5. Signal abort this 's complete or error controller .

6. If error algorithm is not null, then run error algorithm given error .

   Assert : No exception was thrown .

7. Otherwise (i.e., when error algorithm is null), report the exception error .

1. If this 's active is false, then return.

2. If this 's relevant global object is a Window object, and its associated Document is not fully active , then return.

3. Let complete algorithm be this 's complete algorithm .

4. Close this .

5. Signal abort this 's complete or error controller .

6. If complete algorithm is not null, then run complete algorithm .

   Assert : No exception was thrown .

1. If this 's relevant global object is a Window object, and its associated Document is not fully active , then return.

2. If this 's active is true, then append teardown to this 's teardown callbacks list.

3. Otherwise, invoke teardown .

   If an exception E was thrown , then report the exception E .

1. Set subscriber ’s active boolean to false.

2. Set subscriber ’s next algorithm , error algorithm , and complete algorithm all to null.

let innerSubscriber = null;
const producedValues = [];
const controller = new AbortController();
const observable = new Observable(subscriber => {
  innerSubscriber = subscriber;
  subscriber.complete();
});
observable.subscribe({
  next: v => producedValues.push(v),
  complete: () => innerSubscriber.next('from complete'),
  }, {signal: controller.signal}
);
// This invokes the complete() callback, and even though it invokes next() from
// within, the given next() callback will never run, because the subscription
// has already been "closed" before the complete() callback actually executes.
controller.abort();
console.assert(producedValues.length === 0);

1. Set this 's subscribe callback to callback .

   Note: This callback will get invoked later when subscribe() is called.

1. Subscribe to this given observer and options .

1. Report the exception error .

Note: We pull this default out separately so that every place in this specification that natively subscribes to an Observable (i.e., subscribes from spec prose, not going through the subscribe() method) doesn’t have to redundantly define these steps.

Note: We split this algorithm out from the Web IDL subscribe() method, so that spec prose can subscribe to an Observable without going through the Web IDL bindings. See w3c/IntersectionObserver#464 for similar context, where "internal" prose must not go through Web IDL bindings on objects whose properties could be mutated by JavaScript. See § 2.3.3 Promise-returning operators for usage of this.

1. If this 's relevant global object is a Window object, and its associated Document is not fully active , then return.

2. Let internal observer be a new internal observer .

3. Process observer as follows:

   1. If observer is a SubscriptionObserverCallback

      Set internal observer ’s next steps to these steps that take an any value :

      1. Invoke observer with value .

         If an exception E was thrown , then report the exception E .

      If observer is a SubscriptionObserver

      1. If observer ’s next is not null, set internal observer ’s next steps to these steps that take an any value :

         1. Invoke observer ’s next with value .

            If an exception E was thrown , then report the exception E .

      2. If observer ’s error is not null, set internal observer ’s error steps to these steps that take an any error :

         1. Invoke observer ’s error with error .

            If an exception E was thrown , then report the exception E .

      3. If observer ’s complete is not null, set internal observer ’s complete steps to these steps:

         1. Invoke observer ’s complete .

            If an exception E was thrown , then report the exception E .

      If observer is an internal observer

      Set internal observer to observer .

4. Assert : internal observer ’s error steps is either the default error algorithm , or an algorithm that invokes the provided error callback function .

5. Let subscriber be a new Subscriber , initialized as:

   next algorithm

   internal observer ’s next steps

   error algorithm

   internal observer ’s error steps

   complete algorithm

   internal observer ’s complete steps

   signal

   The result of creating a dependent abort signal from the list « subscriber ’s complete or error controller 's signal , options ’s signal if it is non-null», using AbortSignal , and the current realm .

6. If subscriber ’s signal is aborted , then close subscriber .

   Note: This can happen when SubscribeOptions 's signal is already aborted .

7. Otherwise, add the following abort algorithm to subscriber ’s signal :

   1. Close subscriber .

   2. For each teardown of subscriber ’s teardown callbacks sorted in reverse insertion order:

      1. If subscriber ’s relevant global object is a Window object, and its associated Document is not fully active , then abort these steps.

         Note: This step runs repeatedly because each teardown could result in the above Document becoming inactive.

      2. Invoke teardown .

         If an exception E was thrown , call subscriber ’s error() method with E .

8. If this 's subscribe callback is a SubscribeCallback , invoke it with subscriber .

   If an exception E was thrown , call subscriber ’s error() method with E .

9. Otherwise, run the steps given by this 's subscribe callback , given subscriber .

1. Let sourceObservable be this .

2. Let observable be a new Observable whose subscribe callback is an algorithm that takes a Subscriber subscriber and does the following:

   Note that this method involves Subscribing to two Observable s: (1) notifier , and (2) sourceObservable . We "unsubscribe" from both of them in the following situations:

   1. notifier starts emitting values (either "next" or "error"). In this case, we unsubscribe from notifier since we got all we need from it, and no longer need it to keep producing values. We also unsubscribe from sourceObservable , because it no longer needs to produce values that get plumbed through this method’s returned observable , because we’re manually ending the subscription to observable , since notifier finally produced a value.

   2. sourceObservable either error() s or complete() s itself. In this case, we unsubscribe from notifier since we no longer need to listen for values it emits in order to determine when observable can stop mirroring values from sourceObservable (since sourceObservable ran to completion by itself). Unsubscribing from sourceObservable isn’t necessary, since its subscription has been exhausted by itself.

   1. Let notifierObserver be a new internal observer , initialized as follows:

      next steps

      Run subscriber ’s complete() method.

      Note: This will "unsubscribe" from sourceObservable , if it has been subscribed to by this point. This is because sourceObservable is subscribed to with the "outer" subscriber ’s signal as an input signal, and that signal will get aborted when the "outer" subscriber ’s complete() is called above (and below).

      error steps

      Run subscriber ’s complete() method.

      Note: We do not specify complete steps , because if the notifier Observable completes itself, we do not need to complete the subscriber associated with the observable returned from this method. Rather, the observable will continue to mirror sourceObservable uninterrupted.

   2. Let options be a new SubscribeOptions whose signal is subscriber ’s signal .

   3. Subscribe to notifier given notifierObserver and options .

   4. If subscriber ’s active is false, then return.

      Note: This means that sourceObservable ’s subscribe callback will not even get invoked once, if notifier synchronously emits a value. If notifier only "completes" synchronously though (without emitting a "next" or "error" value), then subscriber ’s active will still be true, and we proceed to subscribe to sourceObservable , which observable will mirror uninterrupted.

   5. Let sourceObserver be a new internal observer , initialized as follows:

      next steps

      Run subscriber ’s next() method, given the passed in value .

      error steps

      Run subscriber ’s error() method, given the passed in error .

      complete steps

      Run subscriber ’s complete() method.

      Note: sourceObserver is mostly a pass-through, mirroring everything that sourceObservable emits, with the exception of having the ability to unsubscribe from the notifier Observable in the case where sourceObservable is exhausted before notifier emits anything.

   6. Subscribe to sourceObservable given sourceObserver and options .

3. Return observable .

1. Let sourceObservable be this .

2. Let observable be a new Observable whose subscribe callback is an algorithm that takes a Subscriber subscriber and does the following:

   1. Let idx be an unsigned long long , initially 0.

   2. Let sourceObserver be a new internal observer , initialized as follows:

      next steps

      1. Invoke mapper with the passed in value , and idx , and let mappedValue be the returned value.

         If an exception E was thrown , then run subscriber ’s error() method, given E , and abort these steps.

      2. Increment idx .

      3. Run subscriber ’s next() method, given mappedValue .

      error steps

      Run subscriber ’s error() method, given the passed in error .

      complete steps

      Run subscriber ’s complete() method.

   3. Let options be a new SubscribeOptions whose signal is subscriber ’s signal .

   4. Subscribe to sourceObservable given sourceObserver and options .

3. Return observable .

1. Let sourceObservable be this .

2. Let observable be a new Observable whose subscribe callback is an algorithm that takes a Subscriber subscriber and does the following:

   1. Let sourceObserver be a new internal observer , initialized as follows:

      next steps

      1. Invoke predicate with the passed in value , and let matches be the returned value.

         If an exception E was thrown , then run subscriber ’s error() method, given E , and abort these steps.

      2. If matches is true, then run subscriber ’s next() method, given value .

      error steps

      Run subscriber ’s error() method, given the passed in error .

      complete steps

      Run subscriber ’s complete() method.

   2. Let options be a new SubscribeOptions whose signal is subscriber ’s signal .

   3. Subscribe to sourceObservable given sourceObserver and options .

3. Return observable .

1. Let sourceObservable be this .

2. Let observable be a new Observable whose subscribe callback is an algorithm that takes a Subscriber subscriber and does the following:

   1. If amount is 0, then run subscriber ’s complete() method and abort these steps.

   2. Let sourceObserver be a new internal observer , initialized as follows:

      next steps

      1. Run subscriber ’s next() method with the passed in value .

      2. Decrement amount .

      3. If amount is 0, then run subscriber ’s complete() method.

      error steps

      Run subscriber ’s error() method, given the passed in error .

      complete steps

      Run subscriber ’s complete() method.

   3. Let options be a new SubscribeOptions whose signal is subscriber ’s signal .

   4. Subscribe to sourceObservable given sourceObserver and options .

3. Return observable .

1. Let sourceObservable be this .

2. Let observable be a new Observable whose subscribe callback is an algorithm that takes a Subscriber subscriber and does the following:

   1. Let sourceObserver be a new internal observer , initialized as follows:

      next steps

      1. If amount is > 0, then decrement amount and abort these steps.

      2. Assert : amount is 0.

      3. Run subscriber ’s next() method with the passed in value .

      error steps

      Run subscriber ’s error() method, given the passed in error .

      complete steps

      Run subscriber ’s complete() method.

   2. Let options be a new SubscribeOptions whose signal is subscriber ’s signal .

   3. Subscribe to sourceObservable given sourceObserver and options .

3. Return observable .

1. TODO: Spec this and use mapper .

1. TODO: Spec this and use callback .

1. Let p a new promise .

2. If options ’s signal is not null:

   1. If options ’s signal is aborted , then:

      1. Reject p with options ’s signal 's abort reason .

      2. Return p .

   2. Add the following abort algorithm to options ’s signal :

      1. Reject p with options ’s signal 's abort reason .

      Note: All we have to do here is reject p . Note that the subscription to this Observable will also be canceled automatically, since the "inner" signal (created during subscription ) is a dependent signal of options ’s signal .

3. Let values be a new list .

4. Let observer be a new internal observer , initialized as follows:

   next steps

   Append the passed in value to values .

   error steps

   Reject p with the passed in error .

   complete steps

   Resolve p with values .

5. Subscribe to this given observer and options .

6. Return p .

1. Let p a new promise .

2. Let visitor callback controller be a new AbortController .

3. Let internal options be a new SubscribeOptions whose signal is the result of creating a dependent abort signal from the list « visitor callback controller ’s signal , options ’s signal if non-null», using AbortSignal , and the current realm .

   Many trivial internal observers act as pass-throughs, and do not control the subscription to the Observable that they represent; that is, their error steps and complete steps are called when the subscription is terminated, and their next steps simply pass some version of the given value along the chain.

   For this operator, however, the below observer ’s next steps are responsible for actually aborting the underlying subscription to this , in the event that callback throws an exception. In that case, the SubscribeOptions 's signal we pass through to " Subscribe to an Observable ", needs to be a dependent signal derived from options ’s signal , and the AbortSignal of an AbortController that the next steps below has access to, and can signal abort when needed.

4. If internal options ’s signal is aborted , then:

   1. Reject p with internal options ’s signal 's abort reason .

   2. Return p .

5. Add the following abort algorithm to internal options ’s signal :

   1. Reject p with internal options ’s signal 's abort reason .

      Note: The fact that rejection of p is tied to internal options ’s signal , and not options ’s signal means, that any microtasks queued during the firing of options ’s signal 's abort event will run before p ’s rejection handler runs.

6. Let idx be an unsigned long long , initially 0.

7. Let observer be a new internal observer , initialized as follows:

   next steps

   1. Invoke callback with the passed in value , and idx .

      If an exception E was thrown , then reject p with E , and signal abort visitor callback controller with E .

   2. Increment idx .

   error steps

   Reject p with the passed in error .

   complete steps

   Resolve p with undefined .

8. Subscribe to this given observer and internal options .

9. Return p .

1. TODO: Spec this and use predicate and options .

1. TODO: Spec this and use predicate and options .

1. TODO: Spec this and use predicate and options .

1. TODO: Spec this and use reducer , initialValue , and options .

1. If this 's relevant global object is a Window object, and its associated Document is not fully active , then return.

2. Let event target be this .

3. Let observable be a new Observable , initialized as follows:

   subscribe callback

   An algorithm that takes a Subscriber subscriber and runs these steps:

   1. If event target is null, abort these steps.

      Note: This is meant to capture the fact that event target can be garbage collected by the time this algorithm runs upon subscription.

   2. If subscriber ’s signal is aborted , abort these steps.

   3. Add an event listener with event target and an event listener defined as follows:

      type

      type

      callback

      The result of creating a new Web IDL EventListener instance representing a reference to a function of one argument of type Event event . This function executes the observable event listener invoke algorithm given subscriber and event .

      capture

      options ’s capture

      passive

      options ’s passive

      once

      false

      signal

      null

      Note: The AbortSignal for event listeners added by on() is managed by the Observable itself. See subscribe() and SubscribeOptions .

4. Return observable .

1. Run subscriber ’s next() method with event .