dom/observable/tentative/observable-constructor.any.js

// Because we test that the global error handler is called at various times.
setup({allow_uncaught_exception: true});

test(() => {
  assert_implements(self.Observable, "The Observable interface is not implemented");

  assert_true(
    typeof Observable === "function",
    "Observable constructor is defined"
  );

  assert_throws_js(TypeError, () => { new Observable(); });
}, "Observable constructor");

test(() => {
  let initializerCalled = false;
  const source = new Observable(() => {
    initializerCalled = true;
  });

  assert_false(
    initializerCalled,
    "initializer should not be called by construction"
  );
  source.subscribe();
  assert_true(initializerCalled, "initializer should be called by subscribe");
}, "subscribe() can be called with no arguments");

test(() => {
  assert_implements(self.Subscriber, "The Subscriber interface is not implemented");
  assert_true(
    typeof Subscriber === "function",
    "Subscriber interface is defined as a function"
  );

  assert_throws_js(TypeError, () => { new Subscriber(); });

  let initializerCalled = false;
  new Observable(subscriber => {
    assert_not_equals(subscriber, undefined, "A Subscriber must be passed into the subscribe callback");
    assert_implements(subscriber.next, "A Subscriber object must have a next() method");
    assert_implements(subscriber.complete, "A Subscriber object must have a complete() method");
    assert_implements(subscriber.error, "A Subscriber object must have an error() method");
    initializerCalled = true;
  }).subscribe();
  assert_true(initializerCalled, "initializer should be called by subscribe");
}, "Subscriber interface is not constructible");

test(() => {
  let initializerCalled = false;
  const results = [];

  const source = new Observable((subscriber) => {
    initializerCalled = true;
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
  });

  assert_false(
    initializerCalled,
    "initializer should not be called by construction"
  );

  source.subscribe(x => results.push(x));

  assert_true(initializerCalled, "initializer should be called by subscribe");
  assert_array_equals(
    results,
    [1, 2, 3],
    "should emit values synchronously, but not complete"
  );
}, "Subscribe with just a function as the next handler");

test(() => {
  let initializerCalled = false;
  const results = [];

  const source = new Observable((subscriber) => {
    initializerCalled = true;
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
  });

  assert_false(
    initializerCalled,
    "initializer should not be called by construction"
  );

  source.subscribe({
    next: (x) => results.push(x),
    error: () => assert_unreached("error should not be called"),
    complete: () => results.push("complete"),
  });

  assert_true(initializerCalled, "initializer should be called by subscribe");
  assert_array_equals(
    results,
    [1, 2, 3, "complete"],
    "should emit values synchronously"
  );
}, "Observable constructor calls initializer on subscribe");

test(() => {
  const error = new Error("error");
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.error(error);
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (e) => results.push(e),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(
    results,
    [1, 2, error],
    "should emit error synchronously"
  );
}, "Observable error path called synchronously");

test(() => {
  let subscriber;
  new Observable(s => { subscriber = s }).subscribe();
  const {next, complete, error} = subscriber;
  assert_throws_js(TypeError, () => next(1));
  assert_throws_js(TypeError, () => complete());
  assert_throws_js(TypeError, () => error(1));
}, "Subscriber must have receiver");

test(() => {
  let subscriber;
  new Observable(s => { subscriber = s }).subscribe();
  assert_throws_js(TypeError, () => subscriber.next());
  assert_throws_js(TypeError, () => subscriber.error());
}, "Subscriber next & error must recieve argument");

test(() => {
  let subscriber;
  new Observable(s => { subscriber = s }).subscribe();
  assert_true(subscriber.active);
  assert_false(subscriber.signal.aborted);
  subscriber.complete();
  assert_false(subscriber.active);
  assert_true(subscriber.signal.aborted);
}, "Subscriber complete() will set active to false, and abort signal");

test(() => {
  let subscriber;
  new Observable(s => { subscriber = s }).subscribe();
  assert_true(subscriber.active);
  subscriber.active = false;
  assert_true(subscriber.active);
}, "Subscriber active is readonly");

test(() => {
  let subscriber;
  new Observable(s => { subscriber = s }).subscribe();
  assert_false(subscriber.signal.aborted);
  const oldSignal = subscriber.signal;
  const newSignal = AbortSignal.abort();
  subscriber.signal = newSignal;
  assert_false(subscriber.signal.aborted);
  assert_equals(subscriber.signal, oldSignal, "signal did not change");
}, "Subscriber signal is readonly");

test(() => {
  const error = new Error("error");
  const results = [];
  let errorReported = null;
  let innerSubscriber = null;
  let subscriptionActivityInFinallyAfterThrow;
  let subscriptionActivityInErrorHandlerAfterThrow;

  self.addEventListener("error", e => errorReported = e, {once: true});

  const source = new Observable((subscriber) => {
    innerSubscriber = subscriber;
    subscriber.next(1);
    try {
      throw error;
    } finally {
      subscriptionActivityInFinallyAfterThrow = subscriber.active;
    }
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (e) => {
      subscriptionActivityInErrorHandlerAfterThrow = innerSubscriber.active;
      results.push(e);
    },
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_equals(errorReported, null, "The global error handler should not be " +
      "invoked when the subscribe callback throws an error and the " +
      "subscriber has given an error handler");
  assert_true(subscriptionActivityInFinallyAfterThrow, "Subscriber is " +
      "considered active in finally block before error handler is invoked");
  assert_false(subscriptionActivityInErrorHandlerAfterThrow, "Subscriber is " +
      "considered inactive in error handler block after thrown error");
  assert_array_equals(
    results,
    [1, error],
    "should emit values and the thrown error synchronously"
  );
}, "Observable should error if initializer throws");

test(t => {
  let innerSubscriber = null;
  let activeBeforeComplete = false;
  let activeAfterComplete = false;
  let activeDuringComplete = false;
  let abortedBeforeComplete = false;
  let abortedDuringComplete = false;
  let abortedAfterComplete = false;

  const source = new Observable((subscriber) => {
    innerSubscriber = subscriber;
    activeBeforeComplete = subscriber.active;
    abortedBeforeComplete = subscriber.signal.aborted;

    subscriber.complete();
    activeAfterComplete = subscriber.active;
    abortedAfterComplete = subscriber.signal.aborted;
  });

  source.subscribe({
    complete: () => {
      activeDuringComplete = innerSubscriber.active;
      abortedDuringComplete = innerSubscriber.signal.aborted;
    }
  });
  assert_true(activeBeforeComplete, "Subscription is active before complete");
  assert_false(abortedBeforeComplete, "Subscription is not aborted before complete");
  assert_false(activeDuringComplete,
      "Subscription becomes inactive during Subscriber#complete(), just " +
      "before Observer#complete() callback is invoked");
  assert_true(abortedDuringComplete,
      "Subscription's signal is aborted during Subscriber#complete(), just " +
      "before Observer#complete() callback is invoked");
  assert_false(activeAfterComplete, "Subscription is not active after complete");
  assert_true(abortedAfterComplete, "Subscription is aborted after complete");
}, "Subscription is inactive after complete()");

test(t => {
  let innerSubscriber = null;
  let activeBeforeError = false;
  let activeAfterError = false;
  let activeDuringError = false;
  let abortedBeforeError = false;
  let abortedDuringError = false;
  let abortedAfterError = false;

  const error = new Error("error");
  const source = new Observable((subscriber) => {
    innerSubscriber = subscriber;
    activeBeforeError = subscriber.active;
    abortedBeforeError = subscriber.signal.aborted;

    subscriber.error(error);
    activeAfterError = subscriber.active;
    abortedAfterError = subscriber.signal.aborted;
  });

  source.subscribe({
    error: () => {
      activeDuringError = innerSubscriber.active;
      abortedDuringError = innerSubscriber.signal.aborted;
    }
  });
  assert_true(activeBeforeError, "Subscription is active before error");
  assert_false(abortedBeforeError, "Subscription is not aborted before error");
  assert_false(activeDuringError,
      "Subscription becomes inactive during Subscriber#error(), just " +
      "before Observer#error() callback is invoked");
  assert_true(abortedDuringError,
      "Subscription's signal is aborted during Subscriber#error(), just " +
      "before Observer#error() callback is invoked");
  assert_false(activeAfterError, "Subscription is not active after error");
  assert_true(abortedAfterError, "Subscription is not aborted after error");
}, "Subscription is inactive after error()");

test(t => {
  let innerSubscriber;
  let initialActivity;
  let initialSignalAborted;

  const source = new Observable((subscriber) => {
    innerSubscriber = subscriber;
    initialActivity = subscriber.active;
    initialSignalAborted = subscriber.signal.aborted;
  });

  source.subscribe({}, {signal: AbortSignal.abort('Initially aborted')});
  assert_false(initialActivity);
  assert_true(initialSignalAborted);
  assert_equals(innerSubscriber.signal.reason, 'Initially aborted');
}, "Subscription is inactive when aborted signal is passed in");

test(() => {
  let outerSubscriber = null;

  const source = new Observable(subscriber => outerSubscriber = subscriber);

  const controller = new AbortController();
  source.subscribe({}, {signal: controller.signal});

  assert_not_equals(controller.signal, outerSubscriber.signal);
}, "Subscriber#signal is not the same AbortSignal as the one passed into `subscribe()`");

test(() => {
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.complete();
    subscriber.next(3);
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: () => assert_unreached("error should not be called"),
    complete: () => results.push("complete"),
  });

  assert_array_equals(
    results,
    [1, 2, "complete"],
    "should emit values synchronously, but not nexted values after complete"
  );
}, "Subscription does not emit values after completion");

test(() => {
  const error = new Error("error");
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.error(error);
    subscriber.next(3);
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (e) => results.push(e),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(
    results,
    [1, 2, error],
    "should emit values synchronously, but not nexted values after error"
  );
}, "Subscription does not emit values after error");

test(() => {
  const error = new Error("error");
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.error(error);
    assert_false(subscriber.active, "subscriber is closed after error");
    subscriber.next(3);
    subscriber.complete();
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(results, [1, 2, error], "should emit synchronously");
}, "Completing or nexting a subscriber after an error does nothing");

test(() => {
  const error = new Error("custom error");
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    subscriber.error(error);
  });

  // No error handler provided...
  source.subscribe({
    next: () => assert_unreached("next should not be called"),
    complete: () => assert_unreached("complete should not be called"),
  });

  // ... still the exception is reported to the global.
  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("custom error"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error, "Error object is equivalent");
}, "Errors pushed to the subscriber that are not handled by the subscription " +
   "are reported to the global");

test(() => {
  const error = new Error("custom error");
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    throw error;
  });

  // No error handler provided...
  source.subscribe({
    next: () => assert_unreached("next should not be called"),
    complete: () => assert_unreached("complete should not be called"),
  });

  // ... still the exception is reported to the global.
  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("custom error"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error, "Error object is equivalent");
}, "Errors thrown in the initializer that are not handled by the " +
   "subscription are reported to the global");

test(() => {
  const error = new Error("custom error");
  const results = [];
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.complete();
    subscriber.error(error);
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: () => assert_unreached("error should not be called"),
    complete: () => results.push("complete"),
  });

  assert_array_equals(
    results,
    [1, 2, "complete"],
    "should emit values synchronously, but not error values after complete"
  );

  // Error reporting still happens even after  the subscription is closed.
  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("custom error"),"Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error, "Error object is equivalent");
}, "Subscription reports errors that are pushed after subscriber is closed " +
   "by completion");

test(t => {
  const error = new Error("custom error");
  const results = [];
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.complete();
    throw error;
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: () => assert_unreached("error should not be called"),
    complete: () => results.push("complete"),
  });

  assert_array_equals(results, [1, 2, "complete"],
    "should emit values synchronously, but not error after complete"
  );

  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("custom error"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error, "Error object is equivalent");
}, "Errors thrown by initializer function after subscriber is closed by " +
   "completion are reported");

test(() => {
  const error1 = new Error("error 1");
  const error2 = new Error("error 2");
  const results = [];
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.error(error1);
    throw error2;
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(
    results,
    [1, 2, error1],
    "should emit values synchronously, but not nexted values after error"
  );

  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("error 2"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error2, "Error object is equivalent");
}, "Errors thrown by initializer function after subscriber is closed by " +
   "error are reported");

test(() => {
  const error1 = new Error("error 1");
  const error2 = new Error("error 2");
  const results = [];
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.error(error1);
    subscriber.error(error2);
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(
    results,
    [1, 2, error1],
    "should emit values synchronously, but not nexted values after error"
  );

  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("error 2"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error2, "Error object is equivalent");
}, "Errors pushed by initializer function after subscriber is closed by " +
   "error are reported");

test(() => {
  const results = [];
  const target = new EventTarget();

  const source = new Observable((subscriber) => {
    target.addEventListener('custom event', e => {
      subscriber.next(1);
      subscriber.complete();
      subscriber.error('not a real error');
    });
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => {
      results.push('complete'),
      // Re-entrantly tries to invoke `complete()`. However, this function must
      // only ever run once.
      target.dispatchEvent(new Event('custom event'));
    },
  });

  target.dispatchEvent(new Event('custom event'));

  assert_array_equals(
    results,
    [1, 'complete'],
    "complete() can only be called once, and cannot invoke other Observer methods"
  );
}, "Subscriber#complete() cannot re-entrantly invoke itself");

test(() => {
  const results = [];
  const target = new EventTarget();

  const source = new Observable((subscriber) => {
    target.addEventListener('custom event', e => {
      subscriber.next(1);
      subscriber.error('not a real error');
      subscriber.complete();
    });
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => {
      results.push('error'),
      // Re-entrantly tries to invoke `error()`. However, this function must
      // only ever run once.
      target.dispatchEvent(new Event('custom event'));
    },
    complete: () => results.push('complete'),
  });

  target.dispatchEvent(new Event('custom event'));

  assert_array_equals(
    results,
    [1, 'error'],
    "error() can only be called once, and cannot invoke other Observer methods"
  );
}, "Subscriber#error() cannot re-entrantly invoke itself");

test(() => {
  const results = [];
  let innerSubscriber = null;
  let activeDuringTeardown1 = null;
  let abortedDuringTeardown1 = null;
  let activeDuringTeardown2 = null;
  let abortedDuringTeardown2 = null;

  const source = new Observable((subscriber) => {
    assert_true(subscriber.active);
    assert_false(subscriber.signal.aborted);
    results.push('subscribe() callback');
    innerSubscriber = subscriber;

    subscriber.signal.addEventListener('abort', () => {
      assert_false(subscriber.active);
      assert_true(subscriber.signal.aborted);
      results.push('inner abort handler');
      subscriber.next('next from inner abort handler');
      subscriber.complete();
    });

    subscriber.addTeardown(() => {
      activeDuringTeardown1 = subscriber.active;
      abortedDuringTeardown1 = subscriber.signal.aborted;
      results.push('teardown 1');
    });

    subscriber.addTeardown(() => {
      activeDuringTeardown2 = subscriber.active;
      abortedDuringTeardown2 = subscriber.signal.aborted;
      results.push('teardown 2');
    });
  });

  const ac = new AbortController();
  source.subscribe({
    // This should never get called. If it is, the array assertion below will fail.
    next: (x) => results.push(x),
    complete: () => results.push('complete()')
  }, {signal: ac.signal});

  ac.signal.addEventListener('abort', () => {
    results.push('outer abort handler');
    assert_true(ac.signal.aborted);
    assert_false(innerSubscriber.signal.aborted);
  });

  assert_array_equals(results, ['subscribe() callback']);
  ac.abort();
  results.push('abort() returned');
  // The reason the "inner" abort event handler is invoked first is because the
  // "inner" AbortSignal is not a dependent signal (that would ordinarily get
  // aborted after the parent, aka "outer" signal, is completely finished being
  // aborted). Instead, the order of operations looks like this:
  //   1. "Outer" signal begins to be aborted
  //   2. Its abort algorithms [1] run [2]; the internal abort algorithm here is
  //      the "inner" Subscriber's "Close a subscription" [0].
  //      a. This signals abort on the "inner" Subscriber's signal, firing the
  //         abort event
  //      b. Then, the "inner" Subscriber's teardowns run.
  //   3. Once the "outer" signal's abort algorithms are finished, the abort
  //      event is fired [3], triggering the outer abort handler.
  //
  // [0]: https://wicg.github.io/observable/#close-a-subscription
  // [1]: https://dom.spec.whatwg.org/#abortsignal-abort-algorithms
  // [2]: https://dom.spec.whatwg.org/#ref-for-abortsignal-abort-algorithms%E2%91%A2:~:text=For%20each%20algorithm%20of%20signal%E2%80%99s%20abort%20algorithms%3A%20run%20algorithm
  // [3]: https://dom.spec.whatwg.org/#abortsignal-signal-abort:~:text=Fire%20an%20event%20named%20abort%20at%20signal
  assert_array_equals(results, [
      'subscribe() callback', 'inner abort handler', 'teardown 2', 'teardown 1',
      'outer abort handler', 'abort() returned',
  ]);
  assert_false(activeDuringTeardown1, 'should not be active during teardown callback 1');
  assert_false(activeDuringTeardown2, 'should not be active during teardown callback 2');
  assert_true(abortedDuringTeardown1, 'should be aborted during teardown callback 1');
  assert_true(abortedDuringTeardown2, 'should be aborted during teardown callback 2');
}, "Unsubscription lifecycle");

// In the usual consumer-initiated unsubscription case, when the AbortController
// is aborted after subscription, teardowns run from upstream->downstream. This
// is because for a given Subscriber, when a downstream signal is aborted
// (`ac.signal` in this case), the "Close" algorithm prompts the Subscriber to
// first abort *its* own signal (the one accessible via `Subscriber#signal`) and
// then run its teardowns.
//
// This means upstream Subscribers get the first opportunity their teardowns
// before the control flow is returned to downstream Subscribers to run *their*
// teardowns (after they abort their internal signal).
test(() => {
  const results = [];
  const upstream = new Observable(subscriber => {
    subscriber.signal.addEventListener('abort',
        e => results.push('upstream abort handler'), {once: true});
    subscriber.addTeardown(
        () => results.push(`upstream teardown. reason: ${subscriber.signal.reason}`));
  });
  const middle = new Observable(subscriber => {
    subscriber.signal.addEventListener('abort',
        e => results.push('middle abort handler'), {once: true});
    subscriber.addTeardown(
        () => results.push(`middle teardown. reason: ${subscriber.signal.reason}`));
    upstream.subscribe({}, {signal: subscriber.signal});
  });
  const downstream = new Observable(subscriber => {
    subscriber.signal.addEventListener('abort',
        e => results.push('downstream abort handler'), {once: true});
    subscriber.addTeardown(
        () => results.push(`downstream teardown. reason: ${subscriber.signal.reason}`));
    middle.subscribe({}, {signal: subscriber.signal});
  });

  const ac = new AbortController();
  downstream.subscribe({}, {signal: ac.signal});
  ac.abort('Abort!');
  assert_array_equals(results, [
      'upstream abort handler',
      'upstream teardown. reason: Abort!',
      'middle abort handler',
      'middle teardown. reason: Abort!',
      'downstream abort handler',
      'downstream teardown. reason: Abort!',
  ]);
}, "Teardowns are called in upstream->downstream order on " +
   "consumer-initiated unsubscription");

// This test is like the above, but asserts the exact opposite order of
// teardowns. This is because, since the Subscriber's signal is aborted
// immediately upon construction, `addTeardown()` runs teardowns synchronously
// in subscriber-order, which goes from downstream->upstream.
test(() => {
  const results = [];
  const upstream = new Observable(subscriber => {
    subscriber.addTeardown(
        () => results.push(`upstream teardown. reason: ${subscriber.signal.reason}`));
  });
  const middle = new Observable(subscriber => {
    subscriber.addTeardown(
        () => results.push(`middle teardown. reason: ${subscriber.signal.reason}`));
    upstream.subscribe({}, {signal: subscriber.signal});
  });
  const downstream = new Observable(subscriber => {
    subscriber.addTeardown(
        () => results.push(`downstream teardown. reason: ${subscriber.signal.reason}`));
    middle.subscribe({}, {signal: subscriber.signal});
  });

  downstream.subscribe({}, {signal: AbortSignal.abort('Initial abort')});
  assert_array_equals(results, [
      "downstream teardown. reason: Initial abort",
      "middle teardown. reason: Initial abort",
      "upstream teardown. reason: Initial abort",
  ]);
}, "Teardowns are called in downstream->upstream order on " +
   "consumer-initiated unsubscription with pre-aborted Signal");

// Producer-initiated unsubscription test, capturing the ordering of abort events and teardowns.
test(() => {
  const results = [];

  const source = new Observable(subscriber => {
    subscriber.addTeardown(() => results.push('source teardown'));
    subscriber.signal.addEventListener('abort',
        e => results.push('source abort event'));
  });

  const middle = new Observable(subscriber => {
    subscriber.addTeardown(() => results.push('middle teardown'));
    subscriber.signal.addEventListener('abort',
        e => results.push('middle abort event'));

    source.subscribe(() => {}, {signal: subscriber.signal});
  });

  let innerSubscriber = null;
  const downstream = new Observable(subscriber => {
    innerSubscriber = subscriber;
    subscriber.addTeardown(() => results.push('downstream teardown'));
    subscriber.signal.addEventListener('abort',
        e => results.push('downstream abort event'));

    middle.subscribe(() => {}, {signal: subscriber.signal});
  });

  downstream.subscribe();

  // Trigger a producer-initiated unsubscription from the most-downstream Observable.
  innerSubscriber.complete();

  assert_array_equals(results, [
    'source abort event',
    'source teardown',
    'middle abort event',
    'middle teardown',
    'downstream abort event',
    'downstream teardown',
  ]);
}, "Producer-initiated unsubscription in a downstream Observable fires abort " +
   "events before each teardown, in downstream->upstream order");

test(t => {
  let innerSubscriber = null;
  const source = new Observable(subscriber => {
    innerSubscriber = subscriber;
    subscriber.error('calling error()');
  });

  source.subscribe();
  assert_equals(innerSubscriber.signal.reason, "calling error()",
      "Reason is set correctly");
}, "Subscriber#error() value is stored as Subscriber's AbortSignal's reason");

test(t => {
  const source = new Observable((subscriber) => {
    let n = 0;
    while (!subscriber.signal.aborted) {
      assert_true(subscriber.active);
      subscriber.next(n++);
      if (n > 3) {
        assert_unreached("The subscriber should be closed by now");
      }
    }
    assert_false(subscriber.active);
  });

  const ac = new AbortController();
  const results = [];

  source.subscribe({
    next: (x) => {
      results.push(x);
      if (x === 2) {
        ac.abort();
      }
    },
    error: () => results.push('error'),
    complete: () => results.push('complete')
  }, {signal: ac.signal});

  assert_array_equals(
    results,
    [0, 1, 2],
    "should emit values synchronously before abort"
  );
}, "Aborting a subscription should stop emitting values");

test(() => {
  const error = new Error("custom error");
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable(() => {
    throw error;
  });

  try {
    source.subscribe();
  } catch {
    assert_unreached("subscriber() never throws an error");
  }

  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("custom error"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error, "Error object is equivalent");
}, "Calling subscribe should never throw an error synchronously, initializer throws error");

test(() => {
  const error = new Error("custom error");
  let errorReported = null;

  self.addEventListener("error", e => errorReported = e, { once: true });

  const source = new Observable((subscriber) => {
    subscriber.error(error);
  });

  try {
    source.subscribe();
  } catch {
    assert_unreached("subscriber() never throws an error");
  }

  assert_true(errorReported !== null, "Exception was reported to global");
  assert_true(errorReported.message.includes("custom error"), "Error message matches");
  assert_greater_than(errorReported.lineno, 0, "Error lineno is greater than 0");
  assert_greater_than(errorReported.colno, 0, "Error lineno is greater than 0");
  assert_equals(errorReported.error, error, "Error object is equivalent");
}, "Calling subscribe should never throw an error synchronously, subscriber pushes error");

test(() => {
  let addTeardownCalled = false;
  let activeDuringTeardown;

  const source = new Observable((subscriber) => {
    subscriber.addTeardown(() => {
      addTeardownCalled = true;
      activeDuringTeardown = subscriber.active;
    });
  });

  const ac = new AbortController();
  source.subscribe({}, {signal: ac.signal});

  assert_false(addTeardownCalled, "Teardown is not be called upon subscription");
  ac.abort();
  assert_true(addTeardownCalled, "Teardown is called when subscription is aborted");
  assert_false(activeDuringTeardown, "Teardown observers inactive subscription");
}, "Teardown should be called when subscription is aborted");

test(() => {
  const addTeardownsCalled = [];
  // This is used to snapshot `addTeardownsCalled` from within the subscribe
  // callback, for assertion/comparison later.
  let teardownsSnapshot = [];
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 1"));
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 2"));

    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();

    // We don't run the actual `assert_array_equals` here because if it fails,
    // it won't be properly caught. This is because assertion failures throw an
    // error, and in subscriber callback, thrown errors result in
    // `window.onerror` handlers being called, which this test file doesn't
    // record as an error (see the first line of this file).
    teardownsSnapshot = addTeardownsCalled;
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: () => results.push("unreached"),
    complete: () => results.push("complete"),
  });

  assert_array_equals(
    results,
    [1, 2, 3, "complete"],
    "should emit values and complete synchronously"
  );

  assert_array_equals(teardownsSnapshot, addTeardownsCalled);
  assert_array_equals(addTeardownsCalled, ["teardown 2", "teardown 1"],
      "Teardowns called in LIFO order synchronously after complete()");
}, "Teardowns should be called when subscription is closed by completion");

test(() => {
  const addTeardownsCalled = [];
  let teardownsSnapshot = [];
  const error = new Error("error");
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 1"));
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 2"));

    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.error(error);

    teardownsSnapshot = addTeardownsCalled;
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(
    results,
    [1, 2, 3, error],
    "should emit values and error synchronously"
  );

  assert_array_equals(teardownsSnapshot, addTeardownsCalled);
  assert_array_equals(addTeardownsCalled, ["teardown 2", "teardown 1"],
      "Teardowns called in LIFO order synchronously after error()");
}, "Teardowns should be called when subscription is closed by subscriber pushing an error");

test(() => {
  const addTeardownsCalled = [];
  const error = new Error("error");
  const results = [];

  const source = new Observable((subscriber) => {
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 1"));
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 2"));

    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    throw error;
  });

  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => assert_unreached("complete should not be called"),
  });

  assert_array_equals(
    results,
    [1, 2, 3, error],
    "should emit values and error synchronously"
  );

  assert_array_equals(addTeardownsCalled, ["teardown 2", "teardown 1"],
      "Teardowns called in LIFO order synchronously after thrown error");
}, "Teardowns should be called when subscription is closed by subscriber throwing error");

test(() => {
  const addTeardownsCalled = [];
  const results = [];
  let firstTeardownInvokedSynchronously = false;
  let secondTeardownInvokedSynchronously = false;

  const source = new Observable((subscriber) => {
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 1"));
    if (addTeardownsCalled.length === 1) {
      firstTeardownInvokedSynchronously = true;
    }
    subscriber.addTeardown(() => addTeardownsCalled.push("teardown 2"));
    if (addTeardownsCalled.length === 2) {
      secondTeardownInvokedSynchronously = true;
    }

    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
  });

  const ac = new AbortController();
  ac.abort();
  source.subscribe({
    next: (x) => results.push(x),
    error: (error) => results.push(error),
    complete: () => results.push('complete')
  }, {signal: ac.signal});

  assert_array_equals(results, []);
  assert_true(firstTeardownInvokedSynchronously, "First teardown callback is invoked during addTeardown()");
  assert_true(secondTeardownInvokedSynchronously, "Second teardown callback is invoked during addTeardown()");
  assert_array_equals(addTeardownsCalled, ["teardown 1", "teardown 2"],
      "Teardowns called synchronously upon addition end up in FIFO order");
}, "Teardowns should be called synchronously during addTeardown() if the subscription is inactive");

test(() => {
  const results = [];
  let producerInvocations = 0;
  let teardownInvocations = 0;

  const source = new Observable((subscriber) => {
    producerInvocations++;
    results.push('producer invoked');
    subscriber.addTeardown(() => {
      teardownInvocations++;
      results.push('teardown invoked');
    });
  });

  const ac1 = new AbortController();
  const ac2 = new AbortController();

  // First subscription.
  source.subscribe({}, {signal: ac1.signal});
  assert_equals(producerInvocations, 1,
      "Producer is invoked once for first subscription");

  // Second subscription should reuse the same producer.
  source.subscribe({}, {signal: ac2.signal});
  assert_equals(producerInvocations, 1,
      "Producer should not be invoked again for second subscription");

  // First unsubscribe.
  ac1.abort();
  assert_equals(teardownInvocations, 0,
      "Teardown not run when first subscriber unsubscribes");

  // Second unsubscribe.
  ac2.abort();
  assert_equals(teardownInvocations, 1,
      "Teardown should run after last subscriber unsubscribes");

  assert_array_equals(results, ['producer invoked', 'teardown invoked']);
}, "Multiple subscriptions share the same producer and teardown runs only " +
   "after last subscription abort");

test(() => {
  const results = [];
  let activeSubscriber = null;

  const source = new Observable(subscriber => {
    activeSubscriber = subscriber;
    results.push('producer start');
    subscriber.addTeardown(() => results.push('teardown'));
  });

  // First subscription.
  const ac1 = new AbortController();
  source.subscribe({}, {signal: ac1.signal});
  assert_array_equals(results, ['producer start']);

  // Second subscription.
  const ac2 = new AbortController();
  source.subscribe({}, {signal: ac2.signal});

  // Complete the subscription.
  activeSubscriber.complete();
  assert_array_equals(results, ['producer start', 'teardown']);

  // Additional subscription after complete.
  const ac3 = new AbortController();
  source.subscribe({}, {signal: ac3.signal});

  assert_array_equals(results, ['producer start', 'teardown', 'producer start']);
}, "New subscription after complete creates new producer");

test(() => {
  const results = [];
  let producerInvocations = 0;

  const source = new Observable(subscriber => {
    producerInvocations++;
    results.push('producer start');
    subscriber.addTeardown(() => results.push('teardown'));
  });

  // Create 3 subscriptions.
  const ac1 = new AbortController();
  const ac2 = new AbortController();
  const ac3 = new AbortController();
  source.subscribe({}, {signal: ac1.signal});
  source.subscribe({}, {signal: ac2.signal});
  source.subscribe({}, {signal: ac3.signal});

  assert_equals(producerInvocations, 1, "Producer should be invoked once");

  // Unsubscribe in a different order.
  ac2.abort();
  results.push('after first abort');
  ac1.abort();
  results.push('after second abort');
  ac3.abort();
  results.push('after final abort');

  assert_array_equals(results, [
    'producer start',
    'after first abort',
    'after second abort',
    'teardown',
    'after final abort'
  ]);
}, "Teardown runs after last unsubscribe regardless of unsubscription order");

test(() => {
  const results = [];
  const source = new Observable(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.complete();
  });

  source.subscribe(v => {
    results.push(`${v}-first-sub`);
    if (v === 1) {
      // This new subscription adds a new internal observer to the subscriber's
      // internal observer list, but it does not get the value `1` that the
      // subscriber is currently pushing. This is because the Subscriber
      // iterates over a snapshot of its internal observers to push values to.
      // The first value that this new subscription will see is `2`.
      //
      // See https://github.com/WICG/observable/pull/214.
      source.subscribe(v => results.push(`${v}-second-sub`));
    }
  });

  assert_array_equals(results, [
    "1-first-sub",

    "2-first-sub",
    "2-second-sub",
  ]);
}, "Subscriber iterates over a snapshot of its internal observers");