FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: normalize fused stateless transform results · nodejs/node@5b7d02e · GitHub

/ node Public

Commit 5b7d02e

Browse files
trivikr
authored and
aduh95
committed
stream: normalize fused stateless transform results
Normalize each stateless transform result before passing it to the next transform in a fused run. This ensures that subsequent transforms always receive Uint8Array[] batches in both synchronous and asynchronous pipelines. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: codex:gpt-5.6-sol PR-URL: #65367 Fixes: #65366 Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 8d39d0d commit 5b7d02e

2 files changed

Lines changed: 65 additions & 10 deletions

File tree

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

Lines changed: 33 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -434,6 +434,16 @@ async function appendTransformResultAsyncSlow(target, result) {
434434
}
435435
}
436436

437+
function normalizeTransformResultFast(result) {
438+
if (isUint8ArrayBatch(result)) {
439+
return result.length === 0 ? null : result;
440+
}
441+
if (isUint8Array(result)) return [result];
442+
if (typeof result === 'string') return [toUint8Array(result)];
443+
if (isAnyArrayBuffer(result)) return [new Uint8Array(result)];
444+
if (ArrayBufferIsView(result)) return [arrayBufferViewToUint8Array(result)];
445+
}
446+
437447
// =============================================================================
438448
// Sync Pipeline Implementation
439449
// =============================================================================
@@ -457,7 +467,17 @@ function* applyFusedStatelessSyncTransforms(source, run) {
457467
current = null;
458468
break;
459469
}
460-
current = result;
470+
if (i === run.length - 1) {
471+
current = result;
472+
continue;
473+
}
474+
current = normalizeTransformResultFast(result);
475+
if (current === undefined) {
476+
const normalized = [];
477+
appendTransformResultSync(normalized, result);
478+
current = normalized.length === 0 ? null : normalized[0];
479+
}
480+
if (current === null) break;
461481
}
462482
if (current === null) continue;
463483
// Inline normalization with Uint8Array[] batch as the fast path,
@@ -570,21 +590,24 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
570590
for await (const chunks of source) {
571591
let current = chunks;
572592
for (let i = 0; i < run.length; i++) {
573-
const result = run[i](current, { __proto__: null, signal });
593+
let result = run[i](current, { __proto__: null, signal });
594+
if (isPromise(result)) result = await result;
574595
if (result === null) {
575596
current = null;
576597
break;
577598
}
578-
if (isPromise(result)) {
579-
const resolved = await result;
580-
if (resolved === null) {
581-
current = null;
582-
break;
583-
}
584-
current = resolved;
585-
} else {
599+
if (i === run.length - 1) {
586600
current = result;
601+
continue;
602+
}
603+
current = normalizeTransformResultFast(result);
604+
if (current === undefined) {
605+
const normalized = [];
606+
const pendingResult = appendTransformResultAsync(normalized, result);
607+
if (pendingResult !== undefined) await pendingResult;
608+
current = normalized.length === 0 ? null : normalized[0];
587609
}
610+
if (current === null) break;
588611
}
589612
if (current === null) continue;
590613
// Normalize the final output

‎test/parallel/test-stream-iter-transform-output.js‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,36 @@ async function testSyncTransformReturnsFloat32Array() {
5959
assert.strictEqual(data.byteLength, 4);
6060
}
6161

62+
// Consecutive stateless transforms normalize intermediate output (async)
63+
async function testConsecutiveTransformsNormalizeIntermediateOutput() {
64+
const first = (chunks) => {
65+
return chunks === null ? null : new Uint8Array([65]);
66+
};
67+
let receivedBatch = false;
68+
const second = (chunks) => {
69+
if (chunks !== null) receivedBatch = Array.isArray(chunks);
70+
return chunks;
71+
};
72+
const data = await bytes(pull(from('x'), first, second));
73+
assert.ok(receivedBatch);
74+
assert.deepStrictEqual(data, new Uint8Array([65]));
75+
}
76+
77+
// Consecutive stateless transforms normalize intermediate output (sync)
78+
async function testConsecutiveSyncTransformsNormalizeIntermediateOutput() {
79+
const first = (chunks) => {
80+
return chunks === null ? null : new Uint8Array([65]);
81+
};
82+
let receivedBatch = false;
83+
const second = (chunks) => {
84+
if (chunks !== null) receivedBatch = Array.isArray(chunks);
85+
return chunks;
86+
};
87+
const data = bytesSync(pullSync(fromSync('x'), first, second));
88+
assert.ok(receivedBatch);
89+
assert.deepStrictEqual(data, new Uint8Array([65]));
90+
}
91+
6292
// Stateless transform returns a sync generator (iterable)
6393
async function testTransformReturnsGenerator() {
6494
const tx = (chunks) => {
@@ -233,6 +263,8 @@ Promise.all([
233263
testSyncTransformReturnsArrayBuffer(),
234264
testTransformReturnsFloat32Array(),
235265
testSyncTransformReturnsFloat32Array(),
266+
testConsecutiveTransformsNormalizeIntermediateOutput(),
267+
testConsecutiveSyncTransformsNormalizeIntermediateOutput(),
236268
testTransformReturnsGenerator(),
237269
testSyncTransformReturnsGenerator(),
238270
testTransformReturnsAsyncGenerator(),

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL