diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index c546e70ad96..61dd843e757 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -434,6 +434,16 @@ async function appendTransformResultAsyncSlow(target, result) { } } +function normalizeTransformResultFast(result) { + if (isUint8ArrayBatch(result)) { + return result.length === 0 ? null : result; + } + if (isUint8Array(result)) return [result]; + if (typeof result === 'string') return [toUint8Array(result)]; + if (isAnyArrayBuffer(result)) return [new Uint8Array(result)]; + if (ArrayBufferIsView(result)) return [arrayBufferViewToUint8Array(result)]; +} + // ============================================================================= // Sync Pipeline Implementation // ============================================================================= @@ -457,7 +467,17 @@ function* applyFusedStatelessSyncTransforms(source, run) { current = null; break; } - current = result; + if (i === run.length - 1) { + current = result; + continue; + } + current = normalizeTransformResultFast(result); + if (current === undefined) { + const normalized = []; + appendTransformResultSync(normalized, result); + current = normalized.length === 0 ? null : normalized[0]; + } + if (current === null) break; } if (current === null) continue; // Inline normalization with Uint8Array[] batch as the fast path, @@ -570,21 +590,24 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) { for await (const chunks of source) { let current = chunks; for (let i = 0; i < run.length; i++) { - const result = run[i](current, { __proto__: null, signal }); + let result = run[i](current, { __proto__: null, signal }); + if (isPromise(result)) result = await result; if (result === null) { current = null; break; } - if (isPromise(result)) { - const resolved = await result; - if (resolved === null) { - current = null; - break; - } - current = resolved; - } else { + if (i === run.length - 1) { current = result; + continue; + } + current = normalizeTransformResultFast(result); + if (current === undefined) { + const normalized = []; + const pendingResult = appendTransformResultAsync(normalized, result); + if (pendingResult !== undefined) await pendingResult; + current = normalized.length === 0 ? null : normalized[0]; } + if (current === null) break; } if (current === null) continue; // Normalize the final output diff --git a/test/parallel/test-stream-iter-transform-output.js b/test/parallel/test-stream-iter-transform-output.js index d66a20f6e16..90261a33785 100644 --- a/test/parallel/test-stream-iter-transform-output.js +++ b/test/parallel/test-stream-iter-transform-output.js @@ -59,6 +59,36 @@ async function testSyncTransformReturnsFloat32Array() { assert.strictEqual(data.byteLength, 4); } +// Consecutive stateless transforms normalize intermediate output (async) +async function testConsecutiveTransformsNormalizeIntermediateOutput() { + const first = (chunks) => { + return chunks === null ? null : new Uint8Array([65]); + }; + let receivedBatch = false; + const second = (chunks) => { + if (chunks !== null) receivedBatch = Array.isArray(chunks); + return chunks; + }; + const data = await bytes(pull(from('x'), first, second)); + assert.ok(receivedBatch); + assert.deepStrictEqual(data, new Uint8Array([65])); +} + +// Consecutive stateless transforms normalize intermediate output (sync) +async function testConsecutiveSyncTransformsNormalizeIntermediateOutput() { + const first = (chunks) => { + return chunks === null ? null : new Uint8Array([65]); + }; + let receivedBatch = false; + const second = (chunks) => { + if (chunks !== null) receivedBatch = Array.isArray(chunks); + return chunks; + }; + const data = bytesSync(pullSync(fromSync('x'), first, second)); + assert.ok(receivedBatch); + assert.deepStrictEqual(data, new Uint8Array([65])); +} + // Stateless transform returns a sync generator (iterable) async function testTransformReturnsGenerator() { const tx = (chunks) => { @@ -233,6 +263,8 @@ Promise.all([ testSyncTransformReturnsArrayBuffer(), testTransformReturnsFloat32Array(), testSyncTransformReturnsFloat32Array(), + testConsecutiveTransformsNormalizeIntermediateOutput(), + testConsecutiveSyncTransformsNormalizeIntermediateOutput(), testTransformReturnsGenerator(), testSyncTransformReturnsGenerator(), testTransformReturnsAsyncGenerator(),