tor-browser

observable-forEach.any.js (6184B)

---

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

 const results = [];

 const completion = source.forEach((value) => {
   results.push(value);
 });

 assert_array_equals(results, [1, 2, 3]);
 await completion;
}, "forEach(): Visitor callback called synchronously for each value");

promise_test(async (t) => {
 const error = new Error("error");
 const source = new Observable((subscriber) => {
   throw error;
 });

 try {
   await source.forEach(() => {
     assert_unreached("Visitor callback is not invoked when Observable errors");
   });
   assert_unreached("forEach() promise does not resolve when Observable errors");
 } catch (e) {
   assert_equals(e, error);
 }
}, "Errors thrown by Observable reject the returned promise");

promise_test(async (t) => {
 const error = new Error("error");
 const source = new Observable((subscriber) => {
   subscriber.error(error);
 });

 try {
   await source.forEach(() => {
     assert_unreached("Visitor callback is not invoked when Observable errors");
   });
   assert_unreached("forEach() promise does not resolve when Observable errors");
 } catch (reason) {
   assert_equals(reason, error);
 }
}, "Errors pushed by Observable reject the returned promise");

promise_test(async (t) => {
 // This will be assigned when `source`'s teardown is called during
 // unsubscription.
 let abortReason = null;

 const error = new Error("error");
 const source = new Observable((subscriber) => {
   // Should be called from within the second `next()` call below, when the
   // `forEach()` visitor callback throws an error, because that triggers
   // unsubscription from `source`.
   subscriber.addTeardown(() => abortReason = subscriber.signal.reason);

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

 const results = [];

 const completion = source.forEach((value) => {
   results.push(value);
   if (value === 2) {
     throw error;
   }
 });

 assert_array_equals(results, [1, 2]);
 assert_equals(abortReason, error,
     "forEach() visitor callback throwing an error triggers unsubscription " +
     "from the source observable, with the correct abort reason");

 try {
   await completion;
   assert_unreached("forEach() promise does not resolve when visitor throws");
 } catch (e) {
   assert_equals(e, error);
 }
}, "Errors thrown in the visitor callback reject the promise and " +
  "unsubscribe from the source");

// See https://github.com/WICG/observable/issues/96 for discussion about the
// timing of Observable AbortSignal `abort` firing and promise rejection.
promise_test(async t => {
 const error = new Error('custom error');
 let rejectionError = null;
 let outerAbortEventMicrotaskRun = false,
     forEachPromiseRejectionMicrotaskRun = false,
     innerAbortEventMicrotaskRun = false;

 const source = new Observable(subscriber => {
   subscriber.signal.addEventListener('abort', () => {
     queueMicrotask(() => {
       assert_true(outerAbortEventMicrotaskRun,
           "Inner abort: outer abort microtask has fired");
       assert_true(forEachPromiseRejectionMicrotaskRun,
           "Inner abort: forEach rejection microtask has fired");
       assert_false(innerAbortEventMicrotaskRun,
           "Inner abort: inner abort microtask has not fired");

       innerAbortEventMicrotaskRun = true;
     });
   });
 });

 const controller = new AbortController();
 controller.signal.addEventListener('abort', () => {
   queueMicrotask(() => {
     assert_false(outerAbortEventMicrotaskRun,
         "Outer abort: outer abort microtask has not fired");
     assert_false(forEachPromiseRejectionMicrotaskRun,
         "Outer abort: forEach rejection microtask has not fired");
     assert_false(innerAbortEventMicrotaskRun,
         "Outer abort: inner abort microtask has not fired");

     outerAbortEventMicrotaskRun = true;
   });
 });

 const promise = source.forEach(() => {}, {signal: controller.signal}).catch(e => {
   rejectionError = e;
   assert_true(outerAbortEventMicrotaskRun,
       "Promise rejection: outer abort microtask has fired");
   assert_false(forEachPromiseRejectionMicrotaskRun,
       "Promise rejection: forEach rejection microtask has not fired");
   assert_false(innerAbortEventMicrotaskRun,
       "Promise rejection: inner abort microtask has not fired");

   forEachPromiseRejectionMicrotaskRun = true;
 });

 // This should trigger the following, in this order:
 //   1. Fire the `abort` event at the outer AbortSignal, whose handler
 //      manually queues a microtask.
 //   2. Calls "signal abort" on the outer signal's dependent signals. This
 //      queues a microtask to reject the `forEach()` promise.
 //   3. Fire the `abort` event at the inner AbortSignal, whose handler
 //      manually queues a microtask.
 controller.abort(error);

 // After a single task, assert that everything has happened correctly (and
 // incrementally in the right order);
 await new Promise(resolve => {
   t.step_timeout(resolve);
 });
 assert_true(outerAbortEventMicrotaskRun,
     "Final: outer abort microtask has fired");
 assert_true(forEachPromiseRejectionMicrotaskRun,
     "Final: forEach rejection microtask has fired");
 assert_true(innerAbortEventMicrotaskRun,
     "Final: inner abort microtask has fired");
 assert_equals(rejectionError, error, "Promise is rejected with the right " +
     "value");
}, "forEach visitor callback rejection microtask ordering");

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

 const results = [];

 const completion = source.forEach((value) => {
   results.push(value);
 });

 assert_array_equals(results, [1, 2, 3]);

 const completionValue = await completion;
 assert_equals(completionValue, undefined, "Promise resolves with undefined");
}, "forEach() promise resolves with undefined");