DOM: Implement the forEach() Observable operator

This CL implements the semantics specified in
https://wicg.github.io/observable/#dom-observable-foreach.

See WICG/observable#105.

For WPTs:
Co-authored-by: [email protected]

[email protected]

Bug: 1485981
Change-Id: I61344bad7fa4bac65146e1305a376fc1f5e55dc3
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/5249869
Reviewed-by: Mason Freed <[email protected]>
Commit-Queue: Dominic Farolino <[email protected]>
Cr-Commit-Position: refs/heads/main@{#1257328}

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

@@ -0,0 +1,184 @@
+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");

dom/observable/tentative/observable-forEach.window.js

@@ -0,0 +1,59 @@
+async function loadIframeAndReturnContentWindow() {
+   // Create and attach an iframe.
+  const iframe = document.createElement('iframe');
+  const iframeLoadPromise = new Promise((resolve, reject) => {
+    iframe.onload = resolve;
+    iframe.onerror = reject;
+  });
+  document.body.append(iframe);
+  await iframeLoadPromise;
+  return iframe.contentWindow;
+}
+
+promise_test(async t => {
+  const contentWin = await loadIframeAndReturnContentWindow();
+
+  window.results = [];
+
+  contentWin.eval(`
+    const parentResults = parent.results;
+
+    const source = new Observable(subscriber => {
+      window.frameElement.remove();
+
+      // This invokes the forEach() operator's internal observer's next steps,
+      // which at least in Chromium, must have a special "context is detached"
+      // check to early-return, so as to not crash.
+      subscriber.next(1);
+    });
+
+    source.forEach(value => {
+      parentResults.push(value);
+    });
+  `);
+
+  // If we got here, we didn't crash! Let's also check that `results` is empty.
+  assert_array_equals(results, []);
+}, "forEach()'s internal observer's next steps do not crash in a detached document");
+
+promise_test(async t => {
+  const contentWin = await loadIframeAndReturnContentWindow();
+
+  window.results = [];
+
+  contentWin.eval(`
+    const parentResults = parent.results;
+
+    const source = new Observable(subscriber => {
+      subscriber.next(1);
+    });
+
+    source.forEach(value => {
+      window.frameElement.remove();
+      parentResults.push(value);
+    });
+  `);
+
+  assert_array_equals(results, [1]);
+}, "forEach()'s internal observer's next steps do not crash when visitor " +
+   "callback detaches the document");