stream: support flatMap

Support the `flatMap` method from the iterator helper TC39 proposal on
readable streams.

Co-Authored-By: Robert Nagy <ronagy@icloud.com>
PR-URL: https://github.com/nodejs/node/pull/41612
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

1 files changed, 49 insertions, 0 deletions

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 6ec019875dc..0e763df9655 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2027,6 +2027,55 @@ console.log(allBigFiles);
 console.log('done'); // Stream has finished
 ```
 
+### `readable.flatMap(fn[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `fn` {Function|AsyncGeneratorFunction|AsyncFunction} a function to map over
+  every item in the stream.
+  * `data` {any} a chunk of data from the stream.
+  * `options` {Object}
+    * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
+      abort the `fn` call early.
+* `options` {Object}
+  * `concurrency` {number} the maximum concurrent invocation of `fn` to call
+    on the stream at once. **Default:** `1`.
+  * `signal` {AbortSignal} allows destroying the stream if the signal is
+    aborted.
+* Returns: {Readable} a stream flat-mapped with the function `fn`.
+
+This method returns a new stream by applying the given callback to each
+chunk of the stream and then flattening the result.
+
+It is possible to return a stream or another iterable or async iterable from
+`fn` and the result streams will be merged (flattened) into the returned
+stream.
+
+```mjs
+import { Readable } from 'stream';
+import { createReadStream } from 'fs';
+
+// With a synchronous mapper.
+for await (const item of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
+  console.log(item); // 1, 1, 2, 2, 3, 3, 4, 4
+}
+// With an asynchronous mapper, combine the contents of 4 files
+const concatResult = Readable.from([
+  './1.mjs',
+  './2.mjs',
+  './3.mjs',
+  './4.mjs',
+]).flatMap((fileName) => createReadStream(fileName));
+for await (const result of concatResult) {
+  // This will contain the contents (all chunks) of all 4 files
+  console.log(result);
+}
+```
+
 ### Duplex and transform streams
 
 #### Class: `stream.Duplex`