Skip to content

Commit 0895ba3

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 088841e commit 0895ba3

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 {
@@ -65,10 +61,7 @@ const {
6561
} = require('internal/streams/iter/utils');
6662

6763
const {
68-
kValidatedSource,
6964
kValidatedTransform,
70-
toAsyncStreamable,
71-
toStreamable,
7265
} = require('internal/streams/iter/types');
7366

7467
// =============================================================================
@@ -131,22 +124,6 @@ function parsePipeToArgs(args, requiredMethod) {
131124
};
132125
}
133126

134-
function canUseSyncIterablePipeToFastPath(source, transforms, signal) {
135-
if (signal !== undefined ||
136-
transforms.length !== 0 ||
137-
isPrimitiveChunk(source) ||
138-
ArrayIsArray(source) ||
139-
source?.[kValidatedSource] ||
140-
!isSyncIterable(source) ||
141-
isAsyncIterable(source)) {
142-
return false;
143-
}
144-
145-
// Preserve from()'s top-level protocol precedence for custom iterables.
146-
return typeof source[toAsyncStreamable] !== 'function' &&
147-
typeof source[toStreamable] !== 'function';
148-
}
149-
150127
// =============================================================================
151128
// Transform Output Flattening
152129
// =============================================================================
@@ -1068,9 +1045,7 @@ async function pipeTo(source, ...args) {
10681045
}
10691046

10701047
const hasWriteSync = typeof writer.writeSync === 'function';
1071-
const useSyncIterableFastPath =
1072-
hasWriteSync && canUseSyncIterablePipeToFastPath(source, transforms, signal);
1073-
const normalized = useSyncIterableFastPath ? undefined : from(source);
1048+
const normalized = from(source);
10741049

10751050
let totalBytes = 0;
10761051
const hasWritev = typeof writer.writev === 'function';
@@ -1141,31 +1116,7 @@ async function pipeTo(source, ...args) {
11411116
}
11421117

11431118
try {
1144-
if (useSyncIterableFastPath) {
1145-
// Avoid from()'s async sync-iterable batching path. This keeps writes
1146-
// incremental for synchronous sources while preserving async
1147-
// normalization for non-primitive yielded values.
1148-
for (const value of source) {
1149-
if (isUint8ArrayBatch(value)) {
1150-
if (value.length > 0) {
1151-
const p = writeBatch(value);
1152-
if (p) await p;
1153-
}
1154-
continue;
1155-
}
1156-
if (isUint8Array(value)) {
1157-
const p = writeBatch([value]);
1158-
if (p) await p;
1159-
continue;
1160-
}
1161-
1162-
const batch = await ArrayFromAsync(normalizeAsyncValue(value));
1163-
if (batch.length > 0) {
1164-
const p = writeBatch(batch);
1165-
if (p) await p;
1166-
}
1167-
}
1168-
} else if (transforms.length === 0) {
1119+
if (transforms.length === 0) {
11691120
// Fast path: no transforms - iterate normalized source directly
11701121
if (signal) {
11711122
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
@@ -247,7 +247,7 @@ async function testPipeToSyncMinimalWriter() {
247247
assert.strictEqual(chunks.length > 0, true);
248248
}
249249

250-
async function testPipeToSyncIterableFastPathWritesIncrementally() {
250+
async function testPipeToSyncIterableUsesFromBatching() {
251251
let pulled = 0;
252252
let firstWritePulled = 0;
253253
const chunks = [];
@@ -270,15 +270,39 @@ async function testPipeToSyncIterableFastPathWritesIncrementally() {
270270

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

281-
async function testPipeToSyncIterableFastPathWriteFallback() {
281+
async function testPipeToSourceNormalizationIndependentOfWriter() {
282+
function source() {
283+
return {
284+
*[Symbol.iterator]() {
285+
yield {
286+
async *[Symbol.asyncIterator]() {
287+
yield 'nested';
288+
},
289+
};
290+
},
291+
};
292+
}
293+
294+
for (const hasWriteSync of [false, true]) {
295+
const writer = { write: common.mustNotCall() };
296+
if (hasWriteSync) writer.writeSync = common.mustNotCall();
297+
298+
await assert.rejects(
299+
pipeTo(source(), writer, { preventClose: true, preventFail: true }),
300+
{ code: 'ERR_INVALID_ARG_TYPE' },
301+
);
302+
}
303+
}
304+
305+
async function testPipeToSyncIterableWriteFallback() {
282306
const asyncWrites = [];
283307
const writer = {
284308
writeSync(chunk) {
@@ -299,7 +323,7 @@ async function testPipeToSyncIterableFastPathWriteFallback() {
299323
assert.deepStrictEqual(asyncWrites, [new Uint8Array([0x62])]);
300324
}
301325

302-
async function testPipeToSyncIterableFastPathAsyncValue() {
326+
async function testPipeToSyncIterableAsyncValue() {
303327
const chunks = [];
304328
const writer = {
305329
write: common.mustNotCall(),
@@ -337,7 +361,8 @@ Promise.all([
337361
testPipeToSyncPreventClose(),
338362
testPipeToMinimalWriter(),
339363
testPipeToSyncMinimalWriter(),
340-
testPipeToSyncIterableFastPathWritesIncrementally(),
341-
testPipeToSyncIterableFastPathWriteFallback(),
342-
testPipeToSyncIterableFastPathAsyncValue(),
364+
testPipeToSyncIterableUsesFromBatching(),
365+
testPipeToSyncIterableWriteFallback(),
366+
testPipeToSyncIterableAsyncValue(),
367+
testPipeToSourceNormalizationIndependentOfWriter(),
343368
]).then(common.mustCall());

0 commit comments

Comments
 (0)