DOM: Introduce Subscriber#active attribute

This CL introduces an API, namely `subscriber.active` for determining
the activity of an existing subscription. The `active` attribute gets
set to false by the platform synchronously before the Observer's
completion and error handlers are invoked. This has the effect of being
observably `false` _within_ those handlers, unlike
`subscriber.signal.aborted`. Please see the lengthy design notes in both
WICG/observable#76 and
WICG/observable#82.

R=masonf@chromium.org

For WPTs:
Co-authored-by: ben@benlesh.com

Bug: 1485981
Change-Id: I4604a80347ff30ff7be44a131bfdd3999b71252a

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

@@ -101,38 +101,92 @@ 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);
-    throw error;
-    // TODO(https://github.com/WICG/observable/issues/76): If we add the
-    // `subscriber.closed` attribute, consider a try-finally block to assert
-    // that `subscriber.closed` is true after throwing. Also TODO: ensure that
-    // that would even be the right behavior.
+    try {
+      throw error;
+    } finally {
+      subscriptionActivityInFinallyAfterThrow = subscriber.active;
+    }
   });
 
   source.subscribe({
     next: (x) => results.push(x),
-    error: (e) => results.push(e),
+    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");
 
-// TODO(https://github.com/WICG/observable/issues/76): If we decide the
-// `subscriber.closed` attribute is needed, re-visit these two tests that were
-// originally included:
-// https://github.com/web-platform-tests/wpt/blob/0246526ca46ef4e5eae8b8e4a87dd905c40f5326/dom/observable/tentative/observable-ctor.any.js#L123-L137.
+test(t => {
+  let innerSubscriber = null;
+  let activeAfterComplete = false;
+  let activeDuringComplete = false;
+
+  const source = new Observable((subscriber) => {
+    innerSubscriber = subscriber;
+
+    subscriber.complete();
+    activeAfterComplete = subscriber.active;
+  });
+
+  source.subscribe({complete: () => activeDuringComplete = innerSubscriber.active});
+  assert_false(activeDuringComplete, "Subscription is not active during complete");
+  assert_false(activeAfterComplete, "Subscription is not active after complete");
+}, "Subscription is inactive after complete()");
+
+test(t => {
+  let innerSubscriber = null;
+  let activeAfterError = false;
+  let activeDuringError = false;
+
+  const error = new Error("error");
+  const source = new Observable((subscriber) => {
+    innerSubscriber = subscriber;
+
+    subscriber.error(error);
+    activeAfterError = subscriber.active;
+  });
+
+  source.subscribe({error: () => activeDuringError = innerSubscriber.active});
+  assert_false(activeDuringError, "Subscription is not active during error");
+  assert_false(activeAfterError, "Subscription is not active after error");
+}, "Subscription is inactive after error()");
+
+test(t => {
+  let initialActivity;
+
+  const source = new Observable((subscriber) => {
+    initialActivity = subscriber.active;
+  });
+
+  const ac = new AbortController();
+  ac.abort();
+  source.subscribe({signal: ac.signal});
+  assert_false(initialActivity);
+}, "Subscription is inactive when aborted signal is passed in");
 
 // TODO(domfarolino): Add a test asserting that `Subscriber#signal` != the
 // actual `AbortSignal` passed into `subscribe()`. See