Spec ref-counted producers

spec.bs

@@ -177,8 +177,8 @@ interface Subscriber {
 };
 </xmp>
 
-Each {{Subscriber}} has a <dfn for=Subscriber>internal observer</dfn>, which is an [=internal
-observer=].
+Each {{Subscriber}} has an [=ordered set=] of <dfn for=Subscriber>internal observers</dfn>,
+initially empty.
 
 Each {{Subscriber}} has a <dfn for=Subscriber>teardown callbacks</dfn>, which is a [=list=] of
 {{VoidFunction}}s, initially empty.
@@ -205,21 +205,22 @@ The <dfn attribute for=Subscriber><code>signal</code></dfn> getter steps are to
     1. If [=this=]'s [=relevant global object=] is a {{Window}} object, and its [=associated
        Document=] is not [=Document/fully active=], then return.
 
-    1. Run [=this=]'s [=Subscriber/internal observer=]'s [=internal observer/next steps=] given
-       |value|.
+    1. [=set/For each=] |observer| of [=this=]'s [=Subscriber/internal observers=]:
 
-       [=Assert=]: No <a spec=webidl lt="an exception was thrown">exception was thrown</a>.
+       1. Run |observer|'s [=internal observer/next steps=] given |value|.
 
-       <div class=note>
-         <p>Note: No exception can be thrown here because in the case where the
-         [=Subscriber/internal observer=]'s [=internal observer/next steps=] is just a wrapper
-         around a script-provided callback, the <a href=#process-observer>process observer</a> steps
-         take care to wrap these callbacks in logic that, when invoking them, catches any
-         exceptions, and reports them to the global.</p>
-
-         <p>When the [=internal observer/next steps=] is a spec algorithm, those steps take care to
-         not throw any exceptions outside of itself, to appease this assert.</p>
-       </div>
+          [=Assert=]: No <a spec=webidl lt="an exception was thrown">exception was thrown</a>.
+
+          <div class=note>
+            <p>Note: No exception can be thrown here because in the case where the
+            [=Subscriber/internal observer=]'s [=internal observer/next steps=] is just a wrapper
+            around a script-provided callback, the <a href=#process-observer>process observer</a>
+            steps take care to wrap these callbacks in logic that, when invoking them, catches any
+            exceptions, and reports them to the global.</p>
+
+            <p>When the [=internal observer/next steps=] is a spec algorithm, those steps take care
+            to not throw any exceptions outside of itself, to appease this assert.</p>
+          </div>
 </div>
 
 <div algorithm>
@@ -232,12 +233,13 @@ The <dfn attribute for=Subscriber><code>signal</code></dfn> getter steps are to
 
     1. [=close a subscription|Close=] [=this=].
 
-    1. Run [=this=]'s [=Subscriber/internal observer=]'s [=internal observer/error steps=] given
-       |error|.
+    1. [=set/For each=] |observer| of [=this=]'s [=Subscriber/internal observers=]:
+
+       1. Run |observer|'s [=internal observer/error steps=] given |error|.
 
-       [=Assert=]: No <a spec=webidl lt="an exception was thrown">exception was thrown</a>.
+          [=Assert=]: No <a spec=webidl lt="an exception was thrown">exception was thrown</a>.
 
-       Note: See the documentation in {{Subscriber/next()}} for details on why this is true.
+          Note: See the documentation in {{Subscriber/next()}} for details on why this is true.
 </div>
 
 <div algorithm>
@@ -250,11 +252,13 @@ The <dfn attribute for=Subscriber><code>signal</code></dfn> getter steps are to
 
     1. [=close a subscription|Close=] [=this=].
 
-    1. Run [=this=]'s [=Subscriber/internal observer=]'s [=internal observer/complete steps=].
+    1. [=set/For each=] |observer| of [=this=]'s [=Subscriber/internal observers=]:
 
-       [=Assert=]: No <a spec=webidl lt="an exception was thrown">exception was thrown</a>.
+       1. Run |observer|'s [=internal observer/complete steps=].
 
-       Note: See the documentation in {{Subscriber/next()}} for details on why this is true.
+          [=Assert=]: No <a spec=webidl lt="an exception was thrown">exception was thrown</a>.
+
+          Note: See the documentation in {{Subscriber/next()}} for details on why this is true.
 </div>
 
 <div algorithm>
@@ -407,6 +411,9 @@ interface Observable {
 Each {{Observable}} has a <dfn for=Observable>subscribe callback</dfn>, which is a
 {{SubscribeCallback}} or a set of steps that take in a {{Subscriber}}.
 
+Each {{Observable}} has a <dfn for=Observable>weak subscriber</dfn>, which is a weak reference to a
+{{Subscriber}}-or-null, initially null.
+
 Note: The "union" of these types is to support both {{Observable}}s created by JavaScript (that are
 always constructed with a {{SubscribeCallback}}), and natively-constructed {{Observable}} objects
 (whose [=Observable/subscribe callback=] could be an arbitrary set of native steps, not a JavaScript
@@ -732,9 +739,38 @@ An <dfn>internal observer</dfn> is a [=struct=] with the following [=struct/item
        error algorithm=], or an algorithm that [=invokes=] the provided
        {{SubscriptionObserver/error}} [=callback function=].
 
+    1. If [=this=]'s [=Observable/weak subscriber=] is not null and [=this=]'s [=Observable/weak
+       subscriber=]'s [=Subscriber/active=] is true:
+
+       1. Let |subscriber| be [=this=]'s [=Observable/weak subscriber=].
+
+       1. [=set/Append=] |internal observer| to |subscriber|'s [=Subscriber/internal observers=].
+
+       1. If |options|'s {{SubscribeOptions/signal}} [=map/exists=], then:
+
+          1. If |options|'s {{SubscribeOptions/signal}} is [=AbortSignal/aborted=], then
+             [=set/remove=] |internal observer| from |subscriber|'s [=Subscriber/internal
+             observers=].
+
+          1. Otherwise, [=AbortSignal/add|add the following abort algorithm=] to |options|'s
+             {{SubscribeOptions/signal}}:
+
+             1. If |subscriber|'s [=Subscriber/active=] is false, then abort these steps.
+
+             1. [=set/Remove=] |internal observer| from |subscriber|'s [=Subscriber/internal
+                observers=].
+
+             1. If |subscriber|'s [=Subscriber/internal observers=] is [=set/empty=], then [=close a
+                subscription|close=] |subscriber| with |options|'s {{SubscribeOptions/signal}}'s
+                [=AbortSignal/abort reason=].
+
+       1. Return.
+
     1. Let |subscriber| be a [=new=] {{Subscriber}}.
 
-    1. Set |subscriber|'s [=Subscriber/internal observer=] to |internal observer|.
+    1. [=set/Append=] |internal observer| to |subscriber|'s [=Subscriber/internal observers=].
+
+    1. Set [=this=]'s [=Observable/weak subscriber=] to |subscriber|.
 
     1. If |options|'s {{SubscribeOptions/signal}} [=map/exists=], then:
 
@@ -745,8 +781,14 @@ An <dfn>internal observer</dfn> is a [=struct=] with the following [=struct/item
        1. Otherwise, [=AbortSignal/add|add the following abort algorithm=] to |options|'s
           {{SubscribeOptions/signal}}:
 
-          1. [=close a subscription|Close=] |subscriber| with |options|'s
-             {{SubscribeOptions/signal}} [=AbortSignal/abort reason=].
+          1. If |subscriber|'s [=Subscriber/active=] is false, then abort these steps.
+
+          1. [=set/Remove=] |internal observer| from |subscriber|'s [=Subscriber/internal
+             observers=].
+
+          1. If |subscriber|'s [=Subscriber/internal observers=] is [=set/empty=], then [=close a
+             subscription|close=] |subscriber| with |options|'s {{SubscribeOptions/signal}}'s
+             [=AbortSignal/abort reason=].
 
     1. If [=this=]'s [=Observable/subscribe callback=] is a {{SubscribeCallback}}, [=invoke=] it
        with |subscriber|.