stream: fast-path stateless transform flush results

Avoid generator-based normalization for common flush result types in
fused stateless stream/iter transforms. This keeps the corrected flush
ordering while reducing overhead for already-normalized batches and
primitive byte outputs.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63605
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: trivikr

lib/internal/streams/iter/pull.js

@@ -280,6 +280,31 @@ function* processTransformResultSync(result) {
  * @param {*} result
  */
 function appendTransformResultSync(target, result) {
+  if (result === null) {
+    return;
+  }
+  if (isUint8ArrayBatch(result)) {
+    if (result.length > 0) {
+      ArrayPrototypePush(target, result);
+    }
+    return;
+  }
+  if (isUint8Array(result)) {
+    ArrayPrototypePush(target, [result]);
+    return;
+  }
+  if (typeof result === 'string') {
+    ArrayPrototypePush(target, [toUint8Array(result)]);
+    return;
+  }
+  if (isAnyArrayBuffer(result)) {
+    ArrayPrototypePush(target, [new Uint8Array(result)]);
+    return;
+  }
+  if (ArrayBufferIsView(result)) {
+    ArrayPrototypePush(target, [arrayBufferViewToUint8Array(result)]);
+    return;
+  }
   for (const batch of processTransformResultSync(result)) {
     ArrayPrototypePush(target, batch);
   }
@@ -372,9 +397,38 @@ async function* processTransformResultAsync(result) {
  * Append normalized transform result batches to an array (async).
  * @param {Array<Uint8Array[]>} target
  * @param {*} result
- * @returns {Promise<void>}
+ * @returns {Promise<void>|undefined}
  */
-async function appendTransformResultAsync(target, result) {
+function appendTransformResultAsync(target, result) {
+  if (result === null) {
+    return;
+  }
+  if (isUint8ArrayBatch(result)) {
+    if (result.length > 0) {
+      ArrayPrototypePush(target, result);
+    }
+    return;
+  }
+  if (isUint8Array(result)) {
+    ArrayPrototypePush(target, [result]);
+    return;
+  }
+  if (typeof result === 'string') {
+    ArrayPrototypePush(target, [toUint8Array(result)]);
+    return;
+  }
+  if (isAnyArrayBuffer(result)) {
+    ArrayPrototypePush(target, [new Uint8Array(result)]);
+    return;
+  }
+  if (ArrayBufferIsView(result)) {
+    ArrayPrototypePush(target, [arrayBufferViewToUint8Array(result)]);
+    return;
+  }
+  return appendTransformResultAsyncSlow(target, result);
+}
+
+async function appendTransformResultAsyncSlow(target, result) {
   for await (const batch of processTransformResultAsync(result)) {
     ArrayPrototypePush(target, batch);
   }
@@ -553,13 +607,19 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
   for (let i = 0; i < run.length; i++) {
     const next = [];
     for (let j = 0; j < pending.length; j++) {
-      await appendTransformResultAsync(
+      const pendingResult = appendTransformResultAsync(
         next,
         run[i](pending[j], { __proto__: null, signal }));
+      if (pendingResult !== undefined) {
+        await pendingResult;
+      }
     }
-    await appendTransformResultAsync(
+    const flushResult = appendTransformResultAsync(
       next,
       run[i](null, { __proto__: null, signal }));
+    if (flushResult !== undefined) {
+      await flushResult;
+    }
     pending = next;
   }
   for (let i = 0; i < pending.length; i++) {