stream: more iter streams fixes

doc/api/stream_iter.md

@@ -1374,7 +1374,8 @@ added:
 
 * `input` {AsyncIterable|Iterable|BroadcastChannel}
 * `options` {Object} Same as `broadcast()`.
-* Returns: {Object} `{ writer, broadcast }`
+* Returns: {BroadcastChannel|Object} A `broadcastProtocol` input returns its
+  {BroadcastChannel} directly. Other inputs return `{ writer, broadcast }`.
 
 Create a {BroadcastChannel} from an existing source. The source is consumed
 automatically and pushed to all subscribers.
@@ -1883,7 +1884,7 @@ class MessageBus {
 }
 
 const bus = new MessageBus();
-const { broadcast } = Broadcast.from(bus);
+const broadcast = Broadcast.from(bus);
 const consumer = broadcast.push();
 bus.send('hello');
 bus.close();
@@ -1919,7 +1920,7 @@ class MessageBus {
 }
 
 const bus = new MessageBus();
-const { broadcast } = Broadcast.from(bus);
+const broadcast = Broadcast.from(bus);
 const consumer = broadcast.push();
 bus.send('hello');
 bus.close();

lib/internal/streams/iter/broadcast.js

@@ -10,6 +10,7 @@ const {
   ArrayIsArray,
   ArrayPrototypePush,
   ArrayPrototypeShift,
+  FunctionPrototypeCall,
   PromisePrototypeThen,
   PromiseReject,
   PromiseResolve,
@@ -60,9 +61,9 @@ const {
   kResolvedPromise,
   convertChunks,
   createBatchEntry,
+  getProtocolMethod,
   getWriterSignal,
   getMinCursor,
-  hasProtocol,
   onSignalAbort,
   parsePullArgs,
   toWriterUint8Array,
@@ -87,6 +88,7 @@ const kOnEndDrained = Symbol('kOnEndDrained');
 const kOnCancel = Symbol('kOnCancel');
 const kPendingWriteRemoved = Symbol('kPendingWriteRemoved');
 const kNoBroadcastError = Symbol('kNoBroadcastError');
+const kSetFactorySignal = Symbol('kSetFactorySignal');
 
 function raceEndWithSignal(promise, signal) {
   if (!signal) return promise;
@@ -110,7 +112,7 @@ class BroadcastImpl {
   #buffer = new RingBuffer();
   #bufferStart = 0;
   #consumers = new SafeSet();
-  #waiters = [];  // Consumers with pending resolve (subset of #consumers)
+  #waiters = new SafeSet();  // Consumers with pending resolve
   #ended = false;
   #error;
   #errored = false;
@@ -119,6 +121,7 @@ class BroadcastImpl {
   #writer = null;
   #cachedMinCursor = 0;
   #cachedMinCursorConsumers = 0;
+  #abortHandler;
   /** Cumulative byte size of buffered entries */
   #bufferedBytes = 0;
 
@@ -133,6 +136,11 @@ class BroadcastImpl {
     this.#writer = writer;
   }
 
+  [kSetFactorySignal](signal) {
+    this.#abortHandler = () => this.cancel(signal.reason);
+    onSignalAbort(signal, this.#abortHandler);
+  }
+
   get backpressurePolicy() {
     return this.#options.backpressure;
   }
@@ -203,6 +211,7 @@ class BroadcastImpl {
 
     function detach() {
       state.detached = true;
+      self.#waiters.delete(state);
       if (state.resolve) {
         state.resolve({ __proto__: null, done: true, value: undefined });
       }
@@ -262,7 +271,7 @@ class BroadcastImpl {
             const { promise, resolve, reject } = PromiseWithResolvers();
             state.resolve = resolve;
             state.reject = reject;
-            ArrayPrototypePush(self.#waiters, state);
+            self.#waiters.add(state);
             return promise;
           },
 
@@ -313,7 +322,9 @@ class BroadcastImpl {
       consumer.detached = true;
     }
     this.#consumers.clear();
+    this.#waiters.clear();
     this.#cachedMinCursorConsumers = 0;
+    this.#cleanupFactorySignal();
     const onCancel = this[kOnCancel];
     this[kOnCancel] = null;
     onCancel?.(reason);
@@ -401,6 +412,7 @@ class BroadcastImpl {
         }
       }
     }
+    this.#waiters.clear();
     this.#notifyEndDrained();
   }
 
@@ -422,7 +434,9 @@ class BroadcastImpl {
       consumer.detached = true;
     }
     this.#consumers.clear();
+    this.#waiters.clear();
     this.#cachedMinCursorConsumers = 0;
+    this.#cleanupFactorySignal();
   }
 
   /**
@@ -442,10 +456,18 @@ class BroadcastImpl {
 
   #notifyEndDrained() {
     if (this.#ended && this.#consumers.size === 0) {
+      this.#cleanupFactorySignal();
       this[kOnEndDrained]?.();
     }
   }
 
+  #cleanupFactorySignal() {
+    if (this.#abortHandler !== undefined) {
+      this.#options.signal.removeEventListener('abort', this.#abortHandler);
+      this.#abortHandler = undefined;
+    }
+  }
+
   #recomputeMinCursor() {
     const { minCursor, minCursorConsumers } = getMinCursor(
       this.#consumers, this.#bufferStart + this.#buffer.length);
@@ -489,12 +511,11 @@ class BroadcastImpl {
 
   #notifyConsumers() {
     const waiters = this.#waiters;
-    if (waiters.length === 0) return;
+    if (waiters.size === 0) return;
     // Swap out the waiters list so consumers that re-wait during
     // resolve don't get processed twice in this cycle.
-    this.#waiters = [];
-    for (let i = 0; i < waiters.length; i++) {
-      const consumer = waiters[i];
+    this.#waiters = new SafeSet();
+    for (const consumer of waiters) {
       if (consumer.resolve) {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
@@ -513,11 +534,11 @@ class BroadcastImpl {
           if (consumer.detached && this.#deleteConsumer(consumer)) {
             this.#tryTrimBuffer();
           } else if (this.#promotePending(consumer)) {
-            ArrayPrototypePush(this.#waiters, consumer);
+            this.#waiters.add(consumer);
           }
         } else {
           // Still waiting -- put back
-          ArrayPrototypePush(this.#waiters, consumer);
+          this.#waiters.add(consumer);
         }
       }
     }
@@ -853,10 +874,6 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) {
   signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
 }
 
-function onBroadcastCancel(broadcastImpl, signal) {
-  onSignalAbort(signal, () => broadcastImpl.cancel(signal.reason));
-}
-
 // =============================================================================
 // Public API
 // =============================================================================
@@ -890,26 +907,23 @@ function broadcast(options = { __proto__: null }) {
   broadcastImpl.setWriter(writer);
 
   if (signal) {
-    onBroadcastCancel(broadcastImpl, signal);
+    broadcastImpl[kSetFactorySignal](signal);
   }
 
   return { __proto__: null, writer, broadcast: broadcastImpl };
 }
 
-function isBroadcastable(value) {
-  return hasProtocol(value, broadcastProtocol);
-}
-
 const Broadcast = {
   __proto__: null,
   from(input, options) {
-    if (isBroadcastable(input)) {
-      const bc = input[broadcastProtocol](options);
+    const protocol = getProtocolMethod(input, broadcastProtocol);
+    if (protocol !== undefined) {
+      const bc = FunctionPrototypeCall(protocol, input, options);
       if (bc === null || typeof bc !== 'object') {
         throw new ERR_INVALID_RETURN_VALUE(
           'an object', '[Symbol.for(\'Stream.broadcastProtocol\')]', bc);
       }
-      return { __proto__: null, writer: { __proto__: null }, broadcast: bc };
+      return bc;
     }
 
     const source = from(input);

lib/internal/streams/iter/consumers.js

@@ -15,6 +15,7 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeShift,
   ArrayPrototypeSlice,
+  FunctionPrototypeCall,
   Promise,
   PromisePrototypeThen,
   SafePromiseAllReturnVoid,
@@ -53,6 +54,7 @@ const {
 const {
   concatBytes,
   createBatchEntry,
+  getProtocolMethod,
   validateBatchEntry,
   yieldAbortable,
 } = require('internal/streams/iter/utils');
@@ -391,14 +393,9 @@ function ondrain(drainable) {
     return null;
   }
 
-  if (
-    !(drainableProtocol in drainable) ||
-    typeof drainable[drainableProtocol] !== 'function'
-  ) {
-    return null;
-  }
-
-  return drainable[drainableProtocol]();
+  const protocol = getProtocolMethod(drainable, drainableProtocol);
+  return protocol === undefined ?
+    null : FunctionPrototypeCall(protocol, drainable);
 }
 
 // =============================================================================

lib/internal/streams/iter/duplex.js

@@ -45,8 +45,10 @@ function duplex(options = { __proto__: null }) {
     backpressure: b?.backpressure ?? backpressure,
   });
 
-  const channelA = createDuplexChannel(aWriter, aReadable);
-  const channelB = createDuplexChannel(bWriter, bReadable);
+  let cleanupSignal;
+  const onClose = () => cleanupSignal?.();
+  const channelA = createDuplexChannel(aWriter, aReadable, onClose);
+  const channelB = createDuplexChannel(bWriter, bReadable, onClose);
 
   // Signal handler: fail both writers with the abort reason so consumers
   // see the error. This is an error-path shutdown, not a clean close.
@@ -55,6 +57,11 @@ function duplex(options = { __proto__: null }) {
       const reason = signal.reason;
       aWriter.fail(reason);
       bWriter.fail(reason);
+      cleanupSignal?.();
+    };
+    cleanupSignal = () => {
+      signal.removeEventListener('abort', abortBoth);
+      cleanupSignal = undefined;
     };
     if (signal.aborted) {
       abortBoth();
@@ -67,7 +74,7 @@ function duplex(options = { __proto__: null }) {
   return [channelA, channelB];
 }
 
-function createDuplexChannel(writer, readable) {
+function createDuplexChannel(writer, readable, onClose) {
   // A push readable has one shared consumer state. Keeping an iterator from
   // creation lets close() terminate that state even if no caller has iterated.
   const closeIterator = readable[SymbolAsyncIterator]();
@@ -78,7 +85,7 @@ function createDuplexChannel(writer, readable) {
     get writer() { return writer; },
     get readable() { return readable; },
     close() {
-      closePromise ??= closeDuplexChannel(writer, closeIterator);
+      closePromise ??= closeDuplexChannel(writer, closeIterator, onClose);
       return closePromise;
     },
     [SymbolAsyncDispose]() {
@@ -87,19 +94,23 @@ function createDuplexChannel(writer, readable) {
   };
 }
 
-async function closeDuplexChannel(writer, closeIterator) {
-  const result = writer.endSync();
-  const endPromise = result < 0 ? writer.end() : undefined;
-  const returnPromise = closeIterator.return();
+async function closeDuplexChannel(writer, closeIterator, onClose) {
+  try {
+    const result = writer.endSync();
+    const endPromise = result < 0 ? writer.end() : undefined;
+    const returnPromise = closeIterator.return();
 
-  if (endPromise !== undefined) {
-    try {
-      await SafePromiseAllReturnVoid([endPromise, returnPromise]);
-    } catch (error) {
-      if (!isConsumerReturnError(error)) throw error;
+    if (endPromise !== undefined) {
+      try {
+        await SafePromiseAllReturnVoid([endPromise, returnPromise]);
+      } catch (error) {
+        if (!isConsumerReturnError(error)) throw error;
+      }
+    } else {
+      await returnPromise;
     }
-  } else {
-    await returnPromise;
+  } finally {
+    onClose();
   }
 }