stream: apply more stream/iter spec conformance fixes

Author: jasnell

doc/api/stream_iter.md

@@ -520,15 +520,19 @@ Including the `node:` prefix on the module specifier is optional.
 added: REPLACEME
 -->
 
-* `input` {string|ArrayBuffer|ArrayBufferView|Iterable|AsyncIterable}
+* `input` {string|ArrayBuffer|ArrayBufferView|Iterable|AsyncIterable|Object}
+  Must not be `null` or `undefined`.
 * Returns: {AsyncIterable\<Uint8Array\[]>}
 
 Create an async byte stream from the given input. Strings are UTF-8 encoded.
 `ArrayBuffer` and `ArrayBufferView` values are wrapped as `Uint8Array`. Arrays
 and iterables are recursively flattened and normalized.
 
 Objects implementing `Symbol.for('Stream.toAsyncStreamable')` or
-`Symbol.for('Stream.toStreamable')` are converted via those protocols.
+`Symbol.for('Stream.toStreamable')` are converted via those protocols. The
+`toAsyncStreamable` protocol takes precedence over `toStreamable`, which takes
+precedence over the iteration protocols (`Symbol.asyncIterator`,
+`Symbol.iterator`).
 
 ```mjs
 import { Buffer } from 'node:buffer';
@@ -556,11 +560,15 @@ run().catch(console.error);
 added: REPLACEME
 -->
 
-* `input` {string|ArrayBuffer|ArrayBufferView|Iterable}
+* `input` {string|ArrayBuffer|ArrayBufferView|Iterable|Object}
+  Must not be `null` or `undefined`.
 * Returns: {Iterable\<Uint8Array\[]>}
 
 Synchronous version of [`from()`][]. Returns a sync iterable. Cannot accept
-async iterables or promises.
+async iterables or promises. Objects implementing
+`Symbol.for('Stream.toStreamable')` are converted via that protocol (takes
+precedence over `Symbol.iterator`). The `toAsyncStreamable` protocol is
+ignored entirely.
 
 ```mjs
 import { fromSync, textSync } from 'node:stream/iter';

lib/internal/streams/iter/broadcast.js

@@ -21,9 +21,9 @@ const {
   SafeSet,
   String,
   Symbol,
-   SymbolAsyncDispose,
-   SymbolAsyncIterator,
-   SymbolDispose,
+  SymbolAsyncDispose,
+  SymbolAsyncIterator,
+  SymbolDispose,
   TypedArrayPrototypeGetByteLength,
 } = primordials;
 
@@ -352,7 +352,7 @@ class BroadcastImpl {
   [kCanWrite]() {
     if (this.#ended || this.#cancelled) return false;
     if ((this.#options.backpressure === 'strict' ||
-         this.#options.backpressure === 'block') &&
+        this.#options.backpressure === 'block') &&
         this.#buffer.length >= this.#options.highWaterMark) {
       return false;
     }
@@ -566,15 +566,15 @@ class BroadcastWriter {
     return this.#totalBytes;
   }
 
-   fail(reason) {
-     if (this.#aborted || this.#closed) return;
-     this.#aborted = true;
-     this.#closed = true;
-     const error = reason ?? new ERR_INVALID_STATE('Failed');
-     this.#rejectPendingWrites(error);
-     this.#rejectPendingDrains(error);
-     this.#broadcast[kAbort](error);
-   }
+  fail(reason) {
+    if (this.#aborted || this.#closed) return;
+    this.#aborted = true;
+    this.#closed = true;
+    const error = reason ?? new ERR_INVALID_STATE('Failed');
+    this.#rejectPendingWrites(error);
+    this.#rejectPendingDrains(error);
+    this.#broadcast[kAbort](error);
+  }
 
   [SymbolAsyncDispose]() {
     this.fail();
@@ -632,7 +632,7 @@ class BroadcastWriter {
 
   #resolvePendingWrites() {
     while (this.#pendingWrites.length > 0 &&
-           this.#broadcast[kCanWrite]()) {
+          this.#broadcast[kCanWrite]()) {
       const pending = this.#pendingWrites.shift();
       if (this.#broadcast[kWrite](pending.chunk)) {
         for (let i = 0; i < pending.chunk.length; i++) {
@@ -776,7 +776,7 @@ const Broadcast = {
       } catch (error) {
         const err = isError(error) ? error :
           new ERR_OPERATION_FAILED(String(error));
-         w.fail(err);
+        w.fail(err);
       }
     };
     PromisePrototypeThen(pump(), undefined, () => {});

lib/internal/streams/iter/from.js

@@ -11,11 +11,12 @@ const {
   ArrayPrototypeEvery,
   ArrayPrototypePush,
   ArrayPrototypeSlice,
+  DataViewPrototypeGetBuffer,
+  DataViewPrototypeGetByteLength,
+  DataViewPrototypeGetByteOffset,
   FunctionPrototypeCall,
-  ObjectPrototypeToString,
   SymbolAsyncIterator,
   SymbolIterator,
-  SymbolToPrimitive,
   TypedArrayPrototypeGetBuffer,
   TypedArrayPrototypeGetByteLength,
   TypedArrayPrototypeGetByteOffset,
@@ -30,6 +31,7 @@ const {
 
 const {
   isArrayBuffer,
+  isDataView,
   isPromise,
   isUint8Array,
 } = require('internal/util/types');
@@ -112,27 +114,6 @@ function isAsyncIterable(value) {
   );
 }
 
-/**
- * Check if object has a custom toString() (not Object.prototype.toString).
- * @returns {boolean}
- */
-function hasCustomToString(obj) {
-  const toString = obj.toString;
-  return typeof toString === 'function' &&
-         toString !== ObjectPrototypeToString;
-}
-
-/**
- * Check if object has Symbol.toPrimitive.
- * @returns {boolean}
- */
-function hasToPrimitive(obj) {
-  return (
-    SymbolToPrimitive in obj &&
-    typeof obj[SymbolToPrimitive] === 'function'
-  );
-}
-
 // =============================================================================
 // Primitive Conversion
 // =============================================================================
@@ -156,39 +137,20 @@ function primitiveToUint8Array(chunk) {
     return chunk;
   }
   // Other ArrayBufferView types (Int8Array, DataView, etc.)
+  if (isDataView(chunk)) {
+    return new Uint8Array(
+      DataViewPrototypeGetBuffer(chunk),
+      DataViewPrototypeGetByteOffset(chunk),
+      DataViewPrototypeGetByteLength(chunk),
+    );
+  }
   return new Uint8Array(
     TypedArrayPrototypeGetBuffer(chunk),
     TypedArrayPrototypeGetByteOffset(chunk),
     TypedArrayPrototypeGetByteLength(chunk),
   );
 }
 
-/**
- * Try to coerce an object to string using custom methods.
- * Returns null if object has no custom string coercion.
- * @returns {string|null}
- */
-function tryStringCoercion(obj) {
-  // Check for Symbol.toPrimitive first
-  if (hasToPrimitive(obj)) {
-    const toPrimitive = obj[SymbolToPrimitive];
-    const result = FunctionPrototypeCall(toPrimitive, obj, 'string');
-    if (typeof result === 'string') {
-      return result;
-    }
-    // toPrimitive returned non-string, fall through to toString
-  }
-
-  // Check for custom toString
-  if (hasCustomToString(obj)) {
-    const result = FunctionPrototypeCall(obj.toString, obj);
-    if (typeof result === 'string') {
-      return result;
-    }
-  }
-
-  return null;
-}
 
 // =============================================================================
 // Sync Normalization (for fromSync and sync contexts)
@@ -229,19 +191,10 @@ function* normalizeSyncValue(value) {
     return;
   }
 
-  // Try string coercion for objects with custom toString/toPrimitive
-  if (typeof value === 'object' && value !== null) {
-    const str = tryStringCoercion(value);
-    if (str !== null) {
-      yield toUint8Array(str);
-      return;
-    }
-  }
-
   // Reject: no valid conversion
   throw new ERR_INVALID_ARG_TYPE(
     'value',
-    ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable'],
+    ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'toStreamable'],
     value,
   );
 }
@@ -362,19 +315,11 @@ async function* normalizeAsyncValue(value) {
     return;
   }
 
-  // Try string coercion for objects with custom toString/toPrimitive
-  if (typeof value === 'object' && value !== null) {
-    const str = tryStringCoercion(value);
-    if (str !== null) {
-      yield toUint8Array(str);
-      return;
-    }
-  }
-
   // Reject: no valid conversion
   throw new ERR_INVALID_ARG_TYPE(
     'value',
-    ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'AsyncIterable'],
+    ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'AsyncIterable',
+     'toStreamable', 'toAsyncStreamable'],
     value,
   );
 }
@@ -472,6 +417,10 @@ async function* normalizeAsyncSource(source) {
  * @returns {Iterable<Uint8Array[]>}
  */
 function fromSync(input) {
+  if (input == null) {
+    throw new ERR_INVALID_ARG_TYPE('input', 'a non-null value', input);
+  }
+
   // Check for primitives first (ByteInput)
   if (isPrimitiveChunk(input)) {
     const chunk = primitiveToUint8Array(input);
@@ -518,11 +467,33 @@ function fromSync(input) {
     }
   }
 
+  // Check toStreamable protocol (takes precedence over iteration protocols).
+  // toAsyncStreamable is ignored entirely in fromSync.
+  if (typeof input[toStreamable] === 'function') {
+    return fromSync(input[toStreamable]());
+  }
+
+  // Reject explicit async inputs
+  if (isAsyncIterable(input)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'input',
+      'a synchronous input (not AsyncIterable)',
+      input,
+    );
+  }
+  if (typeof input === 'object' && input !== null && typeof input.then === 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'input',
+      'a synchronous input (not Promise)',
+      input,
+    );
+  }
+
   // Must be a SyncStreamable
   if (!isSyncIterable(input)) {
     throw new ERR_INVALID_ARG_TYPE(
       'input',
-      ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable'],
+      ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'toStreamable'],
       input,
     );
   }
@@ -541,6 +512,10 @@ function fromSync(input) {
  * @returns {AsyncIterable<Uint8Array[]>}
  */
 function from(input) {
+  if (input == null) {
+    throw new ERR_INVALID_ARG_TYPE('input', 'a non-null value', input);
+  }
+
   // Check for primitives first (ByteInput)
   if (isPrimitiveChunk(input)) {
     const chunk = primitiveToUint8Array(input);
@@ -586,11 +561,29 @@ function from(input) {
     }
   }
 
+  // Check toAsyncStreamable protocol (takes precedence over toStreamable and
+  // iteration protocols)
+  if (typeof input[toAsyncStreamable] === 'function') {
+    return {
+      __proto__: null,
+      async *[SymbolAsyncIterator]() {
+        const result = await input[toAsyncStreamable]();
+        yield* from(result)[SymbolAsyncIterator]();
+      },
+    };
+  }
+
+  // Check toStreamable protocol (takes precedence over iteration protocols)
+  if (typeof input[toStreamable] === 'function') {
+    return from(input[toStreamable]());
+  }
+
   // Must be a Streamable (sync or async iterable)
   if (!isSyncIterable(input) && !isAsyncIterable(input)) {
     throw new ERR_INVALID_ARG_TYPE(
       'input',
-      ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'AsyncIterable'],
+      ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable',
+       'AsyncIterable', 'toStreamable', 'toAsyncStreamable'],
       input,
     );
   }