stream: add asIndexedPairs

doc/api/stream.md

@@ -2118,6 +2118,30 @@ import { Readable } from 'stream';
 await Readable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2]
 ```
 
+### `readable.asIndexedPairs([options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `options` {Object}
+ * `signal` {AbortSignal} allows destroying the stream if the signal is
+ aborted.
+* Returns: {Readable} a stream of indexed pairs.
+
+This method returns a new stream with chunks of the underlying stream paired
+with a counter in the form `[index, chunk]`. The first index value is 0 and it
+increases by 1 for each chunk produced.
+
+```mjs
+import { Readable } from 'stream';
+
+const pairs = await Readable.from(['a', 'b', 'c']).asIndexedPairs().toArray();
+console.log(pairs); // [[0, 'a'], [1, 'b'], [2, 'c']]
+```
+
 ### Duplex and transform streams
 
 #### Class: `stream.Duplex`

lib/internal/streams/operators.js

@@ -157,6 +157,16 @@ async function * map(fn, options) {
  }
 }
 
+async function* asIndexedPairs(options) {
+ let index = 0;
+ for await (const val of this) {
+ if (options?.signal?.aborted) {
+ throw new AbortError({ cause: options.signal.reason });
+ }
+ yield [index++, val];
+ }
+}
+
 async function some(fn, options) {
  // https://tc39.es/proposal-iterator-helpers/#sec-iteratorprototype.some
  // Note that some does short circuit but also closes the iterator if it does
@@ -286,6 +296,7 @@ function take(number, options) {
 }
 
 module.exports.streamReturningOperators = {
+ asIndexedPairs,
  drop,
  filter,
  flatMap,

test/parallel/test-stream-asIndexedPairs.mjs

@@ -0,0 +1,47 @@
+import '../common/index.mjs';
+import { Readable } from 'stream';
+import { deepStrictEqual, rejects } from 'assert';
+
+{
+ // asIndexedPairs with a synchronous stream
+ const pairs = await Readable.from([1, 2, 3]).asIndexedPairs().toArray();
+ deepStrictEqual(pairs, [[0, 1], [1, 2], [2, 3]]);
+ const empty = await Readable.from([]).asIndexedPairs().toArray();
+ deepStrictEqual(empty, []);
+}
+
+{
+ // asIndexedPairs works an asynchronous streams
+ const asyncFrom = (...args) => Readable.from(...args).map(async (x) => x);
+ const pairs = await asyncFrom([1, 2, 3]).asIndexedPairs().toArray();
+ deepStrictEqual(pairs, [[0, 1], [1, 2], [2, 3]]);
+ const empty = await asyncFrom([]).asIndexedPairs().toArray();
+ deepStrictEqual(empty, []);
+}
+
+{
+ // Does not enumerate an infinite stream
+ const infinite = () => Readable.from(async function* () {
+ while (true) yield 1;
+ }());
+ const pairs = await infinite().asIndexedPairs().take(3).toArray();
+ deepStrictEqual(pairs, [[0, 1], [1, 1], [2, 1]]);
+ const empty = await infinite().asIndexedPairs().take(0).toArray();
+ deepStrictEqual(empty, []);
+}
+
+{
+ // AbortSignal
+ await rejects(async () => {
+ const ac = new AbortController();
+ const { signal } = ac;
+ const p = Readable.from([1, 2, 3]).asIndexedPairs({ signal }).toArray();
+ ac.abort();
+ await p;
+ }, { name: 'AbortError' });
+
+ await rejects(async () => {
+ const signal = AbortSignal.abort();
+ await Readable.from([1, 2, 3]).asIndexedPairs({ signal }).toArray();
+ }, /AbortError/);
+}

test/parallel/test-windows-failed-heap-allocation.js

@@ -19,8 +19,8 @@ const tmpdir = require('../common/tmpdir');
 tmpdir.refresh();
 
 // --max-old-space-size=3 is the min 'old space' in V8, explodes fast
-const cmd = `"${process.execPath}" --max-old-space-size=3 "${__filename}"`;
-exec(`${cmd} heapBomb`, { cwd: tmpdir.path }, common.mustCall((err) => {
+const cmd = `"${process.execPath}" --max-old-space-size=30 "${__filename}"`;
+exec(`${cmd} heapBomb`, { cwd: tmpdir.path }, common.mustCall((err, stdout, stderr) => {
  const msg = `Wrong exit code of ${err.code}! Expected 134 for abort`;
  // Note: common.nodeProcessAborted() is not asserted here because it
  // returns true on 134 as well as 0x80000003 (V8's base::OS::Abort)