stream: apply more stream/iter conformance fixes

Author: jasnell

lib/internal/streams/iter/consumers.js

@@ -10,7 +10,6 @@
 const {
   ArrayBufferPrototypeGetByteLength,
   ArrayBufferPrototypeSlice,
-  ArrayIsArray,
   ArrayPrototypeFilter,
   ArrayPrototypeMap,
   ArrayPrototypePush,
@@ -42,6 +41,8 @@ const {
 } = require('internal/validators');
 
 const {
+  from,
+  fromSync,
   isAsyncIterable,
   isSyncIterable,
 } = require('internal/streams/iter/from');
@@ -69,11 +70,6 @@ function isMergeOptions(value) {
 
 // Normalize a yielded value to a Uint8Array[] batch. Sources should yield
 // Uint8Array[] but a raw Uint8Array or string is tolerated by wrapping it.
-function ensureBatch(batch) {
-  if (ArrayIsArray(batch)) return batch;
-  return [batch];
-}
-
 // =============================================================================
 // Shared chunk collection helpers
 // =============================================================================
@@ -85,11 +81,12 @@ function ensureBatch(batch) {
  * @returns {Uint8Array[]}
  */
 function collectSync(source, limit) {
+  // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc.
+  const normalized = fromSync(source);
   const chunks = [];
   let totalBytes = 0;
 
-  for (const raw of source) {
-    const batch = ensureBatch(raw);
+  for (const batch of normalized) {
     for (let i = 0; i < batch.length; i++) {
       const chunk = batch[i];
       if (limit !== undefined) {
@@ -115,65 +112,35 @@ function collectSync(source, limit) {
 async function collectAsync(source, signal, limit) {
   signal?.throwIfAborted();
 
+  // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
+  const normalized = from(source);
   const chunks = [];
 
   // Fast path: no signal and no limit
   if (!signal && limit === undefined) {
-    if (isAsyncIterable(source)) {
-      for await (const raw of source) {
-        const batch = ensureBatch(raw);
-        for (let i = 0; i < batch.length; i++) {
-          ArrayPrototypePush(chunks, batch[i]);
-        }
-      }
-    } else if (isSyncIterable(source)) {
-      for (const raw of source) {
-        const batch = ensureBatch(raw);
-        for (let i = 0; i < batch.length; i++) {
-          ArrayPrototypePush(chunks, batch[i]);
-        }
+    for await (const batch of normalized) {
+      for (let i = 0; i < batch.length; i++) {
+        ArrayPrototypePush(chunks, batch[i]);
       }
-    } else {
-      throw new ERR_INVALID_ARG_TYPE('source', ['AsyncIterable', 'Iterable'], source);
     }
     return chunks;
   }
 
   // Slow path: with signal or limit checks
   let totalBytes = 0;
 
-  if (isAsyncIterable(source)) {
-    for await (const raw of source) {
-      const batch = ensureBatch(raw);
-      signal?.throwIfAborted();
-      for (let i = 0; i < batch.length; i++) {
-        const chunk = batch[i];
-        if (limit !== undefined) {
-          totalBytes += TypedArrayPrototypeGetByteLength(chunk);
-          if (totalBytes > limit) {
-            throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
-          }
-        }
-        ArrayPrototypePush(chunks, chunk);
-      }
-    }
-  } else if (isSyncIterable(source)) {
-    for (const raw of source) {
-      const batch = ensureBatch(raw);
-      signal?.throwIfAborted();
-      for (let i = 0; i < batch.length; i++) {
-        const chunk = batch[i];
-        if (limit !== undefined) {
-          totalBytes += TypedArrayPrototypeGetByteLength(chunk);
-          if (totalBytes > limit) {
-            throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
-          }
+  for await (const batch of normalized) {
+    signal?.throwIfAborted();
+    for (let i = 0; i < batch.length; i++) {
+      const chunk = batch[i];
+      if (limit !== undefined) {
+        totalBytes += TypedArrayPrototypeGetByteLength(chunk);
+        if (totalBytes > limit) {
+          throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
         }
-        ArrayPrototypePush(chunks, chunk);
       }
+      ArrayPrototypePush(chunks, chunk);
     }
-  } else {
-    throw new ERR_INVALID_ARG_TYPE('source', ['AsyncIterable', 'Iterable'], source);
   }
 
   return chunks;
@@ -216,7 +183,8 @@ function validateConsumerOptions(options) {
     try {
       new TextDecoder(options.encoding);
     } catch {
-      throw new ERR_INVALID_ARG_VALUE('options.encoding', options.encoding);
+      throw new ERR_INVALID_ARG_VALUE.RangeError(
+        'options.encoding', options.encoding);
     }
   }
 }
@@ -234,7 +202,8 @@ function validateSyncConsumerOptions(options) {
     try {
       new TextDecoder(options.encoding);
     } catch {
-      throw new ERR_INVALID_ARG_VALUE('options.encoding', options.encoding);
+      throw new ERR_INVALID_ARG_VALUE.RangeError(
+        'options.encoding', options.encoding);
     }
   }
 }
@@ -266,7 +235,6 @@ function textSync(source, options = { __proto__: null }) {
   const decoder = new TextDecoder(options.encoding ?? 'utf-8', {
     __proto__: null,
     fatal: true,
-    ignoreBOM: true,
   });
   return decoder.decode(data);
 }
@@ -322,7 +290,6 @@ async function text(source, options = { __proto__: null }) {
   const decoder = new TextDecoder(options.encoding ?? 'utf-8', {
     __proto__: null,
     fatal: true,
-    ignoreBOM: true,
   });
   return decoder.decode(data);
 }

lib/internal/streams/iter/utils.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+  ArrayBufferPrototypeGetByteLength,
   ArrayPrototypeSlice,
   TypedArrayPrototypeGetBuffer,
   TypedArrayPrototypeGetByteLength,
@@ -70,6 +71,21 @@ function allUint8Array(chunks) {
  * @returns {Uint8Array}
  */
 function concatBytes(chunks) {
+  // Empty stream: return zero-length Uint8Array
+  if (chunks.length === 0) {
+    return new Uint8Array(0);
+  }
+  // Single chunk: return directly if buffer is not shared
+  if (chunks.length === 1) {
+    const chunk = chunks[0];
+    const buf = TypedArrayPrototypeGetBuffer(chunk);
+    if (TypedArrayPrototypeGetByteOffset(chunk) === 0 &&
+        TypedArrayPrototypeGetByteLength(chunk) ===
+          ArrayBufferPrototypeGetByteLength(buf)) {
+      return chunk;
+    }
+  }
+  // Multiple chunks or shared buffer: concatenate
   const buf = Buffer.concat(chunks);
   return new Uint8Array(
     TypedArrayPrototypeGetBuffer(buf),

test/parallel/test-stream-iter-consumers-bytes.js

@@ -182,6 +182,23 @@ async function testConsumersNonArrayBatchSync() {
   assert.strictEqual(arr.length, 2);
 }
 
+// Consumers accept string sources directly (normalized via from/fromSync)
+async function testBytesStringSource() {
+  const result = await bytes('hello-bytes');
+  assert.strictEqual(new TextDecoder().decode(result), 'hello-bytes');
+}
+
+function testBytesSyncStringSource() {
+  const result = bytesSync('hello-sync');
+  assert.strictEqual(new TextDecoder().decode(result), 'hello-sync');
+}
+
+async function testTextStringSource() {
+  const { text } = require('stream/iter');
+  const result = await text('direct-string');
+  assert.strictEqual(result, 'direct-string');
+}
+
 Promise.all([
   testBytesSyncBasic(),
   testBytesSyncLimit(),
@@ -197,4 +214,7 @@ Promise.all([
   testArrayAsyncLimit(),
   testConsumersNonArrayBatch(),
   testConsumersNonArrayBatchSync(),
+  testBytesStringSource(),
+  testBytesSyncStringSource(),
+  testTextStringSource(),
 ]).then(common.mustCall());

test/parallel/test-stream-iter-consumers-text.js

@@ -116,6 +116,35 @@ async function testTextMultiChunkSplitCodepoint() {
   assert.strictEqual(result, '€');
 }
 
+// BOM should be stripped (ignoreBOM defaults to false per spec)
+async function testTextBOMStripped() {
+  // UTF-8 BOM: 0xEF, 0xBB, 0xBF followed by 'hi'
+  const withBOM = new Uint8Array([0xEF, 0xBB, 0xBF, 0x68, 0x69]);
+  const result = await text(from(withBOM));
+  assert.strictEqual(result, 'hi');
+}
+
+async function testTextSyncBOMStripped() {
+  const withBOM = new Uint8Array([0xEF, 0xBB, 0xBF, 0x68, 0x69]);
+  const result = textSync(fromSync(withBOM));
+  assert.strictEqual(result, 'hi');
+}
+
+// Unsupported encoding throws RangeError
+async function testTextUnsupportedEncodingThrowsRangeError() {
+  await assert.rejects(
+    () => text(from('hello'), { encoding: 'not-a-real-encoding' }),
+    { name: 'RangeError' },
+  );
+}
+
+function testTextSyncUnsupportedEncodingThrowsRangeError() {
+  assert.throws(
+    () => textSync(fromSync('hello'), { encoding: 'not-a-real-encoding' }),
+    { name: 'RangeError' },
+  );
+}
+
 Promise.all([
   testTextSyncBasic(),
   testTextAsync(),
@@ -128,4 +157,8 @@ Promise.all([
   testTextEmpty(),
   testTextWithSignal(),
   testTextMultiChunkSplitCodepoint(),
+  testTextBOMStripped(),
+  testTextSyncBOMStripped(),
+  testTextUnsupportedEncodingThrowsRangeError(),
+  testTextSyncUnsupportedEncodingThrowsRangeError(),
 ]).then(common.mustCall());