stream: normalize Broadcast.from() byte inputs

Route non-Broadcastable inputs through from(). This makes strings
and ArrayBuffer views byte inputs instead of generic iterables.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #64082
Fixes: #64081
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -44,6 +44,7 @@ const {
 } = require('internal/streams/iter/types');
 
 const {
+  from,
   isAsyncIterable,
   isSyncIterable,
 } = require('internal/streams/iter/from');
@@ -799,7 +800,9 @@ const Broadcast = {
       return { __proto__: null, writer: { __proto__: null }, broadcast: bc };
     }
 
-    if (!isAsyncIterable(input) && !isSyncIterable(input)) {
+    const source = from(input);
+
+    if (!isAsyncIterable(source) && !isSyncIterable(source)) {
       throw new ERR_INVALID_ARG_TYPE(
         'input', ['Broadcastable', 'AsyncIterable', 'Iterable'], input);
     }
@@ -810,8 +813,8 @@ const Broadcast = {
     const pump = async () => {
       const w = result.writer;
       try {
-        if (isAsyncIterable(input)) {
-          for await (const chunks of input) {
+        if (isAsyncIterable(source)) {
+          for await (const chunks of source) {
             signal?.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
@@ -821,8 +824,8 @@ const Broadcast = {
               await w.write(chunks, signal ? { signal } : undefined);
             }
           }
-        } else if (isSyncIterable(input)) {
-          for (const chunks of input) {
+        } else if (isSyncIterable(source)) {
+          for (const chunks of source) {
             signal?.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {

test/parallel/test-stream-iter-broadcast-from.js

@@ -43,6 +43,28 @@ async function testBroadcastFromStringChunks() {
   assert.strictEqual(data, 'foobar');
 }
 
+async function testBroadcastFromStringInput() {
+  const { broadcast: bc } = Broadcast.from('abc');
+  const consumer = bc.push();
+  const data = await text(consumer);
+  assert.strictEqual(data, 'abc');
+}
+
+async function testBroadcastFromUint8ArrayInput() {
+  const { broadcast: bc } = Broadcast.from(new Uint8Array([97]));
+  const consumer = bc.push();
+  const data = await text(consumer);
+  assert.strictEqual(data, 'a');
+}
+
+async function testBroadcastFromDataViewInput() {
+  const view = new DataView(new Uint8Array([104, 105]).buffer);
+  const { broadcast: bc } = Broadcast.from(view);
+  const consumer = bc.push();
+  const data = await text(consumer);
+  assert.strictEqual(data, 'hi');
+}
+
 async function testBroadcastFromMultipleConsumers() {
   const source = from('shared-data');
   const { broadcast: bc } = Broadcast.from(source);
@@ -180,6 +202,9 @@ Promise.all([
   testBroadcastFromAsyncIterable(),
   testBroadcastFromNonArrayChunks(),
   testBroadcastFromStringChunks(),
+  testBroadcastFromStringInput(),
+  testBroadcastFromUint8ArrayInput(),
+  testBroadcastFromDataViewInput(),
   testBroadcastFromMultipleConsumers(),
   testAbortSignal(),
   testAlreadyAbortedSignal(),

test/parallel/test-stream-iter-validation.js

@@ -157,9 +157,8 @@ assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG
   writer.endSync();
 }
 
-// Broadcast.from rejects non-iterable input
+// Broadcast.from rejects non-streamable input
 assert.throws(() => Broadcast.from(42), { code: 'ERR_INVALID_ARG_TYPE' });
-assert.throws(() => Broadcast.from('bad'), { code: 'ERR_INVALID_ARG_TYPE' });
 
 // =============================================================================
 // share() / shareSync() validation