stream: make multiple updates, cleanups, and fixes

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -12,7 +12,6 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeSlice,
   MathMax,
-  Promise,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseWithResolvers,
@@ -29,6 +28,7 @@ const { isError, lazyDOMException } = require('internal/util');
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_RETURN_VALUE,
     ERR_INVALID_STATE,
   },
 } = require('internal/errors');
@@ -81,16 +81,17 @@ function isPushStreamOptions(value) {
 
 function parsePushArgs(args) {
   if (args.length === 0) {
-    return { transforms: [], options: undefined };
+    return { __proto__: null, transforms: [], options: undefined };
   }
   const last = args[args.length - 1];
   if (isPushStreamOptions(last)) {
     return {
+      __proto__: null,
       transforms: ArrayPrototypeSlice(args, 0, -1),
       options: last,
     };
   }
-  return { transforms: args, options: undefined };
+  return { __proto__: null, transforms: args, options: undefined };
 }
 
 // =============================================================================
@@ -140,7 +141,7 @@ class BroadcastImpl {
     if (transforms.length > 0) {
       if (options?.signal) {
         return pullWithTransforms(
-          rawConsumer, ...transforms, { signal: options.signal });
+          rawConsumer, ...transforms, { __proto__: null, signal: options.signal });
       }
       return pullWithTransforms(rawConsumer, ...transforms);
     }
@@ -149,6 +150,7 @@ class BroadcastImpl {
 
   #createRawConsumer() {
     const state = {
+      __proto__: null,
       cursor: this.#bufferStart + this.#buffer.length,
       resolve: null,
       reject: null,
@@ -399,9 +401,9 @@ class BroadcastWriter {
     const desired = this.desiredSize;
     if (desired === null) return null;
     if (desired > 0) return PromiseResolve(true);
-    return new Promise((resolve, reject) => {
-      ArrayPrototypePush(this.#pendingDrains, { resolve, reject });
-    });
+    const { promise, resolve, reject } = PromiseWithResolvers();
+    ArrayPrototypePush(this.#pendingDrains, { __proto__: null, resolve, reject });
+    return promise;
   }
 
   get desiredSize() {
@@ -442,10 +444,8 @@ class BroadcastWriter {
   async #writevSlow(chunks, options) {
     const signal = options?.signal;
 
-    // Check for pre-aborted signal
-    if (signal?.aborted) {
-      throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-    }
+    // Check for pre-aborted
+    signal?.throwIfAborted();
 
     if (this.#closed || this.#aborted) {
       throw new ERR_INVALID_STATE('Writer is closed');
@@ -558,35 +558,37 @@ class BroadcastWriter {
    * Create a pending write promise, optionally racing against a signal.
    * If the signal fires, the entry is removed from pendingWrites and the
    * promise rejects. Signal listeners are cleaned up on normal resolution.
+   * @returns {Promise<void>}
    */
   #createPendingWrite(chunk, signal) {
-    return new Promise((resolve, reject) => {
-      const entry = { chunk, resolve, reject };
-      this.#pendingWrites.push(entry);
-
-      if (!signal) return;
-
-      const onAbort = () => {
-        // Remove from queue so it doesn't occupy a slot
-        const idx = this.#pendingWrites.indexOf(entry);
-        if (idx !== -1) this.#pendingWrites.removeAt(idx);
-        reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
-      };
-
-      // Wrap resolve/reject to clean up signal listener
-      const origResolve = entry.resolve;
-      const origReject = entry.reject;
-      entry.resolve = function() {
-        signal.removeEventListener('abort', onAbort);
-        origResolve();
-      };
-      entry.reject = function(reason) {
-        signal.removeEventListener('abort', onAbort);
-        origReject(reason);
-      };
-
-      signal.addEventListener('abort', onAbort, { once: true });
-    });
+    const { promise, resolve, reject } = PromiseWithResolvers();
+    const entry = { __proto__: null, chunk, resolve, reject };
+    this.#pendingWrites.push(entry);
+
+    if (!signal) return promise;
+
+    const onAbort = () => {
+      // Remove from queue so it doesn't occupy a slot
+      const idx = this.#pendingWrites.indexOf(entry);
+      if (idx !== -1) this.#pendingWrites.removeAt(idx);
+      reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
+    };
+
+    // Wrap resolve/reject to clean up signal listener
+    const origResolve = entry.resolve;
+    const origReject = entry.reject;
+    entry.resolve = function() {
+      signal.removeEventListener('abort', onAbort);
+      origResolve();
+    };
+    entry.reject = function(reason) {
+      signal.removeEventListener('abort', onAbort);
+      origReject(reason);
+    };
+
+    signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
+
+    return promise;
   }
 
   #resolvePendingWrites() {
@@ -639,6 +641,7 @@ class BroadcastWriter {
  */
 function broadcast(options) {
   const opts = {
+    __proto__: null,
     highWaterMark: MathMax(1, options?.highWaterMark ?? 16),
     backpressure: options?.backpressure ?? 'strict',
     signal: options?.signal,
@@ -654,11 +657,11 @@ function broadcast(options) {
     } else {
       opts.signal.addEventListener('abort', () => {
         broadcastImpl.cancel();
-      }, { once: true });
+      }, { __proto__: null, once: true });
     }
   }
 
-  return { writer, broadcast: broadcastImpl };
+  return { __proto__: null, writer, broadcast: broadcastImpl };
 }
 
 function isBroadcastable(value) {
@@ -674,7 +677,11 @@ const Broadcast = {
   from(input, options) {
     if (isBroadcastable(input)) {
       const bc = input[broadcastProtocol](options);
-      return { writer: {}, broadcast: bc };
+      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 };
     }
 
     const result = broadcast(options);
@@ -685,10 +692,7 @@ const Broadcast = {
       try {
         if (isAsyncIterable(input)) {
           for await (const chunks of input) {
-            if (signal?.aborted) {
-              throw signal.reason ??
-                lazyDOMException('Aborted', 'AbortError');
-            }
+            signal?.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
                 await w.writev(chunks, signal ? { signal } : undefined);
@@ -699,10 +703,7 @@ const Broadcast = {
           }
         } else if (isSyncIterable(input)) {
           for (const chunks of input) {
-            if (signal?.aborted) {
-              throw signal.reason ??
-                lazyDOMException('Aborted', 'AbortError');
-            }
+            signal?.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
                 await w.writev(chunks, signal ? { signal } : undefined);

lib/internal/streams/iter/consumers.js

@@ -33,7 +33,6 @@ const {
   },
 } = require('internal/errors');
 const { TextDecoder } = require('internal/encoding');
-const { lazyDOMException } = require('internal/util');
 
 const {
   isAsyncIterable,
@@ -107,9 +106,7 @@ function collectSync(source, limit) {
  * @returns {Promise<Uint8Array[]>}
  */
 async function collectAsync(source, signal, limit) {
-  if (signal?.aborted) {
-    throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-  }
+  signal?.throwIfAborted();
 
   const chunks = [];
 
@@ -141,9 +138,7 @@ async function collectAsync(source, signal, limit) {
   if (isAsyncIterable(source)) {
     for await (const raw of source) {
       const batch = ensureBatch(raw);
-      if (signal?.aborted) {
-        throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-      }
+      signal?.throwIfAborted();
       for (let i = 0; i < batch.length; i++) {
         const chunk = batch[i];
         if (limit !== undefined) {
@@ -158,9 +153,7 @@ async function collectAsync(source, signal, limit) {
   } else if (isSyncIterable(source)) {
     for (const raw of source) {
       const batch = ensureBatch(raw);
-      if (signal?.aborted) {
-        throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-      }
+      signal?.throwIfAborted();
       for (let i = 0; i < batch.length; i++) {
         const chunk = batch[i];
         if (limit !== undefined) {
@@ -219,6 +212,7 @@ function bytesSync(source, options) {
 function textSync(source, options) {
   const data = bytesSync(source, options);
   const decoder = new TextDecoder(options?.encoding ?? 'utf-8', {
+    __proto__: null,
     fatal: true,
     ignoreBOM: true,
   });
@@ -269,6 +263,7 @@ async function bytes(source, options) {
 async function text(source, options) {
   const data = await bytes(source, options);
   const decoder = new TextDecoder(options?.encoding ?? 'utf-8', {
+    __proto__: null,
     fatal: true,
     ignoreBOM: true,
   });
@@ -376,20 +371,17 @@ function merge(...args) {
   }
 
   return {
+    __proto__: null,
     async *[SymbolAsyncIterator]() {
       const signal = options?.signal;
 
-      if (signal?.aborted) {
-        throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-      }
+      signal?.throwIfAborted();
 
       if (sources.length === 0) return;
 
       if (sources.length === 1) {
         for await (const batch of sources[0]) {
-          if (signal?.aborted) {
-            throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-          }
+          signal?.throwIfAborted();
           yield batch;
         }
         return;
@@ -404,6 +396,7 @@ function merge(...args) {
           // Wrap sync iterator to async
           const syncIter = source[SymbolIterator]();
           iterator = {
+            __proto__: null,
             next() { return PromiseResolve(syncIter.next()); },
             return() {
               return PromiseResolve(syncIter.return?.() ??
@@ -414,7 +407,7 @@ function merge(...args) {
           throw new ERR_INVALID_ARG_TYPE(
             'source', ['AsyncIterable', 'Iterable'], source);
         }
-        return { iterator, done: false, pending: null };
+        return { __proto__: null, iterator, done: false, pending: null };
       });
 
       const startIterator = (state, index) => {
@@ -432,9 +425,7 @@ function merge(...args) {
 
       try {
         while (true) {
-          if (signal?.aborted) {
-            throw signal.reason ?? lazyDOMException('Aborted', 'AbortError');
-          }
+          signal?.throwIfAborted();
 
           const pending = ArrayPrototypeFilter(
             ArrayPrototypeMap(states,

lib/internal/streams/iter/duplex.js

@@ -40,6 +40,7 @@ function duplex(options) {
   let bWriterRef = bWriter;
 
   const channelA = {
+    __proto__: null,
     get writer() { return aWriter; },
     readable: aReadable,
     async close() {
@@ -56,6 +57,7 @@ function duplex(options) {
   };
 
   const channelB = {
+    __proto__: null,
     get writer() { return bWriter; },
     readable: bReadable,
     async close() {

lib/internal/streams/iter/from.js

@@ -468,6 +468,7 @@ function fromSync(input) {
   if (isPrimitiveChunk(input)) {
     const chunk = primitiveToUint8Array(input);
     return {
+      __proto__: null,
       *[SymbolIterator]() {
         yield [chunk];
       },
@@ -482,6 +483,7 @@ function fromSync(input) {
   if (ArrayIsArray(input)) {
     if (input.length === 0) {
       return {
+        __proto__: null,
         *[SymbolIterator]() {
           // Empty - yield nothing
         },
@@ -493,6 +495,7 @@ function fromSync(input) {
       if (allUint8) {
         const batch = input;
         return {
+          __proto__: null,
           *[SymbolIterator]() {
             if (batch.length <= FROM_BATCH_SIZE) {
               yield batch;
@@ -517,6 +520,7 @@ function fromSync(input) {
   }
 
   return {
+    __proto__: null,
     *[SymbolIterator]() {
       yield* normalizeSyncSource(input);
     },
@@ -533,6 +537,7 @@ function from(input) {
   if (isPrimitiveChunk(input)) {
     const chunk = primitiveToUint8Array(input);
     return {
+      __proto__: null,
       async *[SymbolAsyncIterator]() {
         yield [chunk];
       },
@@ -547,6 +552,7 @@ function from(input) {
   if (ArrayIsArray(input)) {
     if (input.length === 0) {
       return {
+        __proto__: null,
         async *[SymbolAsyncIterator]() {
           // Empty - yield nothing
         },
@@ -557,6 +563,7 @@ function from(input) {
       if (allUint8) {
         const batch = input;
         return {
+          __proto__: null,
           async *[SymbolAsyncIterator]() {
             if (batch.length <= FROM_BATCH_SIZE) {
               yield batch;
@@ -581,6 +588,7 @@ function from(input) {
   }
 
   return {
+    __proto__: null,
     async *[SymbolAsyncIterator]() {
       yield* normalizeAsyncSource(input);
     },