Commit f162234
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>1 parent af91029 commit f162234
3 files changed
Lines changed: 34 additions & 7 deletions
File tree
- lib/internal/streams/iter
- test/parallel
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -44,6 +44,7 @@ const { | |||
| 44 | 44 | } = require('internal/streams/iter/types'); | |
| 45 | 45 | ||
| 46 | 46 | const { | |
| 47 | + from, | ||
| 47 | 48 | isAsyncIterable, | |
| 48 | 49 | isSyncIterable, | |
| 49 | 50 | } = require('internal/streams/iter/from'); | |
@@ -799,7 +800,9 @@ const Broadcast = { | |||
| 799 | 800 | return { __proto__: null, writer: { __proto__: null }, broadcast: bc }; | |
| 800 | 801 | } | |
| 801 | 802 | ||
| 802 | - if (!isAsyncIterable(input) && !isSyncIterable(input)) { | ||
| 803 | + const source = from(input); | ||
| 804 | + | ||
| 805 | + if (!isAsyncIterable(source) && !isSyncIterable(source)) { | ||
| 803 | 806 | throw new ERR_INVALID_ARG_TYPE( | |
| 804 | 807 | 'input', ['Broadcastable', 'AsyncIterable', 'Iterable'], input); | |
| 805 | 808 | } | |
@@ -810,8 +813,8 @@ const Broadcast = { | |||
| 810 | 813 | const pump = async () => { | |
| 811 | 814 | const w = result.writer; | |
| 812 | 815 | try { | |
| 813 | - if (isAsyncIterable(input)) { | ||
| 814 | - for await (const chunks of input) { | ||
| 816 | + if (isAsyncIterable(source)) { | ||
| 817 | + for await (const chunks of source) { | ||
| 815 | 818 | signal?.throwIfAborted(); | |
| 816 | 819 | if (ArrayIsArray(chunks)) { | |
| 817 | 820 | if (!w.writevSync(chunks)) { | |
@@ -821,8 +824,8 @@ const Broadcast = { | |||
| 821 | 824 | await w.write(chunks, signal ? { signal } : undefined); | |
| 822 | 825 | } | |
| 823 | 826 | } | |
| 824 | - } else if (isSyncIterable(input)) { | ||
| 825 | - for (const chunks of input) { | ||
| 827 | + } else if (isSyncIterable(source)) { | ||
| 828 | + for (const chunks of source) { | ||
| 826 | 829 | signal?.throwIfAborted(); | |
| 827 | 830 | if (ArrayIsArray(chunks)) { | |
| 828 | 831 | if (!w.writevSync(chunks)) { | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -43,6 +43,28 @@ async function testBroadcastFromStringChunks() { | |||
| 43 | 43 | assert.strictEqual(data, 'foobar'); | |
| 44 | 44 | } | |
| 45 | 45 | ||
| 46 | + async function testBroadcastFromStringInput() { | ||
| 47 | + const { broadcast: bc } = Broadcast.from('abc'); | ||
| 48 | + const consumer = bc.push(); | ||
| 49 | + const data = await text(consumer); | ||
| 50 | + assert.strictEqual(data, 'abc'); | ||
| 51 | + } | ||
| 52 | + | ||
| 53 | + async function testBroadcastFromUint8ArrayInput() { | ||
| 54 | + const { broadcast: bc } = Broadcast.from(new Uint8Array([97])); | ||
| 55 | + const consumer = bc.push(); | ||
| 56 | + const data = await text(consumer); | ||
| 57 | + assert.strictEqual(data, 'a'); | ||
| 58 | + } | ||
| 59 | + | ||
| 60 | + async function testBroadcastFromDataViewInput() { | ||
| 61 | + const view = new DataView(new Uint8Array([104, 105]).buffer); | ||
| 62 | + const { broadcast: bc } = Broadcast.from(view); | ||
| 63 | + const consumer = bc.push(); | ||
| 64 | + const data = await text(consumer); | ||
| 65 | + assert.strictEqual(data, 'hi'); | ||
| 66 | + } | ||
| 67 | + | ||
| 46 | 68 | async function testBroadcastFromMultipleConsumers() { | |
| 47 | 69 | const source = from('shared-data'); | |
| 48 | 70 | const { broadcast: bc } = Broadcast.from(source); | |
@@ -180,6 +202,9 @@ Promise.all([ | |||
| 180 | 202 | testBroadcastFromAsyncIterable(), | |
| 181 | 203 | testBroadcastFromNonArrayChunks(), | |
| 182 | 204 | testBroadcastFromStringChunks(), | |
| 205 | + testBroadcastFromStringInput(), | ||
| 206 | + testBroadcastFromUint8ArrayInput(), | ||
| 207 | + testBroadcastFromDataViewInput(), | ||
| 183 | 208 | testBroadcastFromMultipleConsumers(), | |
| 184 | 209 | testAbortSignal(), | |
| 185 | 210 | 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 | |
0 commit comments