stream: normalize Broadcast.from() byte inputs

trivikr · richardlau · commit f162234f2480 · 2026-07-03T16:05:12.000Z
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>
diff --git a/lib/internal/streams/iter/broadcast.js b/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)) {
diff --git a/test/parallel/test-stream-iter-broadcast-from.js b/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(),
diff --git a/test/parallel/test-stream-iter-validation.js b/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

-Original file line number
+Diff line change
 }=require('internal/streams/iter/types');
 const{
 + from,
  isAsyncIterable,
  isSyncIterable,
 }=require('internal/streams/iter/from');
 return{__proto__: null,writer: {__proto__: null},broadcast: bc};
+}
 -if(!isAsyncIterable(input)&&!isSyncIterable(input)){
 +constsource=from(input);
++
 +if(!isAsyncIterable(source)&&!isSyncIterable(source)){
 thrownewERR_INVALID_ARG_TYPE(
 'input',['Broadcastable','AsyncIterable','Iterable'],input);
+}
 constpump=async()=>{
 constw=result.writer;
 try{
 -if(isAsyncIterable(input)){
 -forawait(constchunksofinput){
 +if(isAsyncIterable(source)){
 +forawait(constchunksofsource){
 signal?.throwIfAborted();
 if(ArrayIsArray(chunks)){
 if(!w.writevSync(chunks)){
 awaitw.write(chunks,signal ? { signal } : undefined);
+}
+}
 -}elseif(isSyncIterable(input)){
 -for(constchunksofinput){
 +}elseif(isSyncIterable(source)){
 +for(constchunksofsource){
 signal?.throwIfAborted();
 if(ArrayIsArray(chunks)){
 if(!w.writevSync(chunks)){
-Original file line number
+Diff line change
 assert.strictEqual(data,'foobar');
+}
 +asyncfunctiontestBroadcastFromStringInput(){
 +const{broadcast: bc}=Broadcast.from('abc');
 +constconsumer=bc.push();
 +constdata=awaittext(consumer);
 +assert.strictEqual(data,'abc');
 +}
++
 +asyncfunctiontestBroadcastFromUint8ArrayInput(){
 +const{broadcast: bc}=Broadcast.from(newUint8Array([97]));
 +constconsumer=bc.push();
 +constdata=awaittext(consumer);
 +assert.strictEqual(data,'a');
 +}
++
 +asyncfunctiontestBroadcastFromDataViewInput(){
 +constview=newDataView(newUint8Array([104,105]).buffer);
 +const{broadcast: bc}=Broadcast.from(view);
 +constconsumer=bc.push();
 +constdata=awaittext(consumer);
 +assert.strictEqual(data,'hi');
 +}
++
 asyncfunctiontestBroadcastFromMultipleConsumers(){
 constsource=from('shared-data');
 const{broadcast: bc}=Broadcast.from(source);
 testBroadcastFromAsyncIterable(),
 testBroadcastFromNonArrayChunks(),
 testBroadcastFromStringChunks(),
 +testBroadcastFromStringInput(),
 +testBroadcastFromUint8ArrayInput(),
 +testBroadcastFromDataViewInput(),
 testBroadcastFromMultipleConsumers(),
 testAbortSignal(),
 testAlreadyAbortedSignal(),
Original file line number	Diff line number	Diff line change
`@@ -157,9 +157,8 @@ assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG`
`157`	`157`	`writer.endSync();`
`158`	`158`	`}`
`159`	`159`
`160`		`-// Broadcast.from rejects non-iterable input`
	`160`	`+// Broadcast.from rejects non-streamable input`
`161`	`161`	`assert.throws(()=>Broadcast.from(42),{code: 'ERR_INVALID_ARG_TYPE'});`
`162`		`-assert.throws(()=>Broadcast.from('bad'),{code: 'ERR_INVALID_ARG_TYPE'});`
`163`	`162`
`164`	`163`	`// =============================================================================`
`165`	`164`	`// share() / shareSync() validation`