Skip to content

Commit 49ee85b

Browse files
jasnelladuh95
jasnell
authored and
aduh95
committed
stream: make pipeTo source normalization independent of Writer
Signed-off-by: James M Snell <[email protected]> Assisted-by: Opencode PR-URL: #65658 Reviewed-By: Trivikram Kamat <[email protected]>
1 parent d7d0440 commit 49ee85b

2 files changed

Lines changed: 34 additions & 58 deletions

File tree

‎lib/internal/streams/iter/pull.js‎

Lines changed: 2 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,6 @@
88

99
const {
1010
ArrayBufferIsView,
11-
ArrayFromAsync,
12-
ArrayIsArray,
1311
ArrayPrototypePush,
1412
ArrayPrototypeSlice,
1513
FunctionPrototypeCall,
@@ -46,9 +44,7 @@ const {
4644
fromSync,
4745
isSyncIterable,
4846
isAsyncIterable,
49-
isPrimitiveChunk,
5047
isUint8ArrayBatch,
51-
normalizeAsyncValue,
5248
} = require('internal/streams/iter/from');
5349

5450
const {
@@ -62,10 +58,7 @@ const {
6258
} = require('internal/streams/iter/utils');
6359

6460
const {
65-
kValidatedSource,
6661
kValidatedTransform,
67-
toAsyncStreamable,
68-
toStreamable,
6962
} = require('internal/streams/iter/types');
7063

7164
// =============================================================================
@@ -128,22 +121,6 @@ function parsePipeToArgs(args, requiredMethod) {
128121
};
129122
}
130123

131-
function canUseSyncIterablePipeToFastPath(source, transforms, signal) {
132-
if (signal !== undefined ||
133-
transforms.length !== 0 ||
134-
isPrimitiveChunk(source) ||
135-
ArrayIsArray(source) ||
136-
source?.[kValidatedSource] ||
137-
!isSyncIterable(source) ||
138-
isAsyncIterable(source)) {
139-
return false;
140-
}
141-
142-
// Preserve from()'s top-level protocol precedence for custom iterables.
143-
return typeof source[toAsyncStreamable] !== 'function' &&
144-
typeof source[toStreamable] !== 'function';
145-
}
146-
147124
// =============================================================================
148125
// Transform Output Flattening
149126
// =============================================================================
@@ -1055,9 +1032,7 @@ async function pipeTo(source, ...args) {
10551032
}
10561033

10571034
const hasWriteSync = typeof writer.writeSync === 'function';
1058-
const useSyncIterableFastPath =
1059-
hasWriteSync && canUseSyncIterablePipeToFastPath(source, transforms, signal);
1060-
const normalized = useSyncIterableFastPath ? undefined : from(source);
1035+
const normalized = from(source);
10611036

10621037
let totalBytes = 0;
10631038
const hasWritev = typeof writer.writev === 'function';
@@ -1118,31 +1093,7 @@ async function pipeTo(source, ...args) {
11181093
}
11191094

11201095
try {
1121-
if (useSyncIterableFastPath) {
1122-
// Avoid from()'s async sync-iterable batching path. This keeps writes
1123-
// incremental for synchronous sources while preserving async
1124-
// normalization for non-primitive yielded values.
1125-
for (const value of source) {
1126-
if (isUint8ArrayBatch(value)) {
1127-
if (value.length > 0) {
1128-
const p = writeBatch(value);
1129-
if (p) await p;
1130-
}
1131-
continue;
1132-
}
1133-
if (isUint8Array(value)) {
1134-
const p = writeBatch([value]);
1135-
if (p) await p;
1136-
continue;
1137-
}
1138-
1139-
const batch = await ArrayFromAsync(normalizeAsyncValue(value));
1140-
if (batch.length > 0) {
1141-
const p = writeBatch(batch);
1142-
if (p) await p;
1143-
}
1144-
}
1145-
} else if (transforms.length === 0) {
1096+
if (transforms.length === 0) {
11461097
// Fast path: no transforms - iterate normalized source directly
11471098
if (signal) {
11481099
for await (const batch of yieldAbortable(normalized, signal)) {

‎test/parallel/test-stream-iter-pipeto.js‎

Lines changed: 32 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -244,7 +244,7 @@ async function testPipeToSyncMinimalWriter() {
244244
assert.strictEqual(chunks.length > 0, true);
245245
}
246246

247-
async function testPipeToSyncIterableFastPathWritesIncrementally() {
247+
async function testPipeToSyncIterableUsesFromBatching() {
248248
let pulled = 0;
249249
let firstWritePulled = 0;
250250
const chunks = [];
@@ -267,15 +267,39 @@ async function testPipeToSyncIterableFastPathWritesIncrementally() {
267267

268268
const totalBytes = await pipeTo(source(), writer);
269269
assert.strictEqual(totalBytes, 3);
270-
assert.strictEqual(firstWritePulled, 1);
270+
assert.strictEqual(firstWritePulled, 3);
271271
assert.deepStrictEqual(chunks, [
272272
new Uint8Array([0x61]),
273273
new Uint8Array([0x62]),
274274
new Uint8Array([0x63]),
275275
]);
276276
}
277277

278-
async function testPipeToSyncIterableFastPathWriteFallback() {
278+
async function testPipeToSourceNormalizationIndependentOfWriter() {
279+
function source() {
280+
return {
281+
*[Symbol.iterator]() {
282+
yield {
283+
async *[Symbol.asyncIterator]() {
284+
yield 'nested';
285+
},
286+
};
287+
},
288+
};
289+
}
290+
291+
for (const hasWriteSync of [false, true]) {
292+
const writer = { write: common.mustNotCall() };
293+
if (hasWriteSync) writer.writeSync = common.mustNotCall();
294+
295+
await assert.rejects(
296+
pipeTo(source(), writer, { preventClose: true, preventFail: true }),
297+
{ code: 'ERR_INVALID_ARG_TYPE' },
298+
);
299+
}
300+
}
301+
302+
async function testPipeToSyncIterableWriteFallback() {
279303
const asyncWrites = [];
280304
const writer = {
281305
writeSync(chunk) {
@@ -296,7 +320,7 @@ async function testPipeToSyncIterableFastPathWriteFallback() {
296320
assert.deepStrictEqual(asyncWrites, [new Uint8Array([0x62])]);
297321
}
298322

299-
async function testPipeToSyncIterableFastPathAsyncValue() {
323+
async function testPipeToSyncIterableAsyncValue() {
300324
const chunks = [];
301325
const writer = {
302326
write: common.mustNotCall(),
@@ -334,7 +358,8 @@ Promise.all([
334358
testPipeToSyncPreventClose(),
335359
testPipeToMinimalWriter(),
336360
testPipeToSyncMinimalWriter(),
337-
testPipeToSyncIterableFastPathWritesIncrementally(),
338-
testPipeToSyncIterableFastPathWriteFallback(),
339-
testPipeToSyncIterableFastPathAsyncValue(),
361+
testPipeToSyncIterableUsesFromBatching(),
362+
testPipeToSyncIterableWriteFallback(),
363+
testPipeToSyncIterableAsyncValue(),
364+
testPipeToSourceNormalizationIndependentOfWriter(),
340365
]).then(common.mustCall());

0 commit comments

Comments
 (0)