[pull] main from nodejs:main

doc/api/stream.md

@@ -69,6 +69,11 @@
 <!-- YAML
 added: v15.0.0
 changes:
+  - version:
+      - v19.7.0
+      - v18.16.0
+    pr-url: https://github.com/nodejs/node/pull/46307
+    description: Added support for webstreams.
   - version:
       - v18.0.0
       - v17.2.0
@@ -79,13 +84,14 @@
                  ends.
 -->
 
-* `streams` {Stream\[]|Iterable\[]|AsyncIterable\[]|Function\[]}
-* `source` {Stream|Iterable|AsyncIterable|Function}
+* `streams` {Stream\[]|Iterable\[]|AsyncIterable\[]|Function\[]|
+  ReadableStream\[]|WritableStream\[]|TransformStream\[]}
+* `source` {Stream|Iterable|AsyncIterable|Function|ReadableStream}
   * Returns: {Promise|AsyncIterable}
-* `...transforms` {Stream|Function}
+* `...transforms` {Stream|Function|TransformStream}
   * `source` {AsyncIterable}
   * Returns: {Promise|AsyncIterable}
-* `destination` {Stream|Function}
+* `destination` {Stream|Function|WritableStream}
   * `source` {AsyncIterable}
   * Returns: {Promise|AsyncIterable}
 * `options` {Object} Pipeline options

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++) {