fixup! stream: add iterator helper find

doc/api/stream.md

@@ -1745,7 +1745,8 @@ added: v17.4.0
 
 > Stability: 1 - Experimental
 
-* `fn` {Function|AsyncFunction} a function to map over every item in the stream.
+* `fn` {Function|AsyncFunction} a function to map over every chunk in the
+  stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1758,16 +1759,16 @@ added: v17.4.0
 * Returns: {Readable} a stream mapped with the function `fn`.
 
 This method allows mapping over the stream. The `fn` function will be called
-for every item in the stream. If the `fn` function returns a promise - that
+for every chunk in the stream. If the `fn` function returns a promise - that
 promise will be `await`ed before being passed to the result stream.
 
 ```mjs
 import { Readable } from 'stream';
 import { Resolver } from 'dns/promises';
 
 // With a synchronous mapper.
-for await (const item of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
-  console.log(item); // 2, 4, 6, 8
+for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
+  console.log(chunk); // 2, 4, 6, 8
 }
 // With an asynchronous mapper, making at most 2 queries at a time.
 const resolver = new Resolver();
@@ -1789,7 +1790,7 @@ added: v17.4.0
 
 > Stability: 1 - Experimental
 
-* `fn` {Function|AsyncFunction} a function to filter items from stream.
+* `fn` {Function|AsyncFunction} a function to filter chunks from the stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1801,8 +1802,8 @@ added: v17.4.0
     aborted.
 * Returns: {Readable} a stream filtered with the predicate `fn`.
 
-This method allows filtering the stream. For each item in the stream the `fn`
-function will be called and if it returns a truthy value, the item will be
+This method allows filtering the stream. For each chunk in the stream the `fn`
+function will be called and if it returns a truthy value, the chunk will be
 passed to the result stream. If the `fn` function returns a promise - that
 promise will be `await`ed.
 
@@ -1811,8 +1812,8 @@ import { Readable } from 'stream';
 import { Resolver } from 'dns/promises';
 
 // With a synchronous predicate.
-for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
-  console.log(item); // 3, 4
+for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
+  console.log(chunk); // 3, 4
 }
 // With an asynchronous predicate, making at most 2 queries at a time.
 const resolver = new Resolver();
@@ -1838,7 +1839,7 @@ added: REPLACEME
 
 > Stability: 1 - Experimental
 
-* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
+* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1850,12 +1851,12 @@ added: REPLACEME
     aborted.
 * Returns: {Promise} a promise for when the stream has finished.
 
-This method allows iterating a stream. For each item in the stream the
+This method allows iterating a stream. For each chunk in the stream the
 `fn` function will be called. If the `fn` function returns a promise - that
 promise will be `await`ed.
 
 This method is different from `for await...of` loops in that it can optionally
-process items concurrently. In addition, a `forEach` iteration can only be
+process chunks concurrently. In addition, a `forEach` iteration can only be
 stopped by having passed a `signal` option and aborting the related
 `AbortController` while `for await...of` can be stopped with `break` or
 `return`. In either case the stream will be destroyed.
@@ -1869,8 +1870,8 @@ import { Readable } from 'stream';
 import { Resolver } from 'dns/promises';
 
 // With a synchronous predicate.
-for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
-  console.log(item); // 3, 4
+for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
+  console.log(chunk); // 3, 4
 }
 // With an asynchronous predicate, making at most 2 queries at a time.
 const resolver = new Resolver();
@@ -1935,7 +1936,7 @@ added: REPLACEME
 
 > Stability: 1 - Experimental
 
-* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
+* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1984,7 +1985,7 @@ added: REPLACEME
 
 > Stability: 1 - Experimental
 
-* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
+* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2000,7 +2001,7 @@ added: REPLACEME
 This method is similar to `Array.prototype.find` and calls `fn` on each chunk
 in the stream to find a chunk with a truthy value for `fn`. Once an `fn` call's
 awaited return value is truthy, the stream is destroyed and the promise is
-fulfilled with value for which the value of `fn` was truthy for. If all of the
+fulfilled with value for which `fn` returned a truthy value. If all of the
 `fn` calls on the chunks return a falsy value, the promise is fulfilled with
 `undefined`.
 
@@ -2034,7 +2035,7 @@ added: REPLACEME
 
 > Stability: 1 - Experimental
 
-* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
+* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2084,7 +2085,7 @@ added: REPLACEME
 > Stability: 1 - Experimental
 
 * `fn` {Function|AsyncGeneratorFunction|AsyncFunction} a function to map over
-  every item in the stream.
+  every chunk in the stream.
   * `data` {any} a chunk of data from the stream.
   * `options` {Object}
     * `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2108,8 +2109,8 @@ 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
+for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
+  console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
 }
 // With an asynchronous mapper, combine the contents of 4 files
 const concatResult = Readable.from([

test/parallel/test-stream-filter.js

@@ -98,3 +98,11 @@ const { setTimeout } = require('timers/promises');
   const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => true);
   assert.strictEqual(stream.readable, true);
 }
+{
+  const stream = Readable.from([1, 2, 3, 4, 5]);
+  Object.defineProperty(stream, 'map', {
+    value: common.mustNotCall(() => {}),
+  });
+  // Check that map isn't getting called.
+  stream.filter(() => true);
+}

test/parallel/test-stream-flatMap.js

@@ -121,3 +121,11 @@ function oneTo5() {
   const stream = oneTo5().flatMap((x) => x);
   assert.strictEqual(stream.readable, true);
 }
+{
+  const stream = oneTo5();
+  Object.defineProperty(stream, 'map', {
+    value: common.mustNotCall(() => {}),
+  });
+  // Check that map isn't getting called.
+  stream.flatMap(() => true);
+}

test/parallel/test-stream-forEach.js

@@ -84,3 +84,11 @@ const { setTimeout } = require('timers/promises');
   const stream = Readable.from([1, 2, 3, 4, 5]).forEach((_) => true);
   assert.strictEqual(typeof stream.then, 'function');
 }
+{
+  const stream = Readable.from([1, 2, 3, 4, 5]);
+  Object.defineProperty(stream, 'map', {
+    value: common.mustNotCall(() => {}),
+  });
+  // Check that map isn't getting called.
+  stream.forEach(() => true);
+}

test/parallel/test-stream-some-every.js → test/parallel/test-stream-some-find-every.mjs

@@ -1,11 +1,8 @@
-'use strict';
+import * as common from '../common/index.mjs';
+import { setTimeout } from 'timers/promises';
+import { Readable } from 'stream';
+import assert from 'assert';
 
-const common = require('../common');
-const { setTimeout } = require('timers/promises');
-const {
-  Readable,
-} = require('stream');
-const assert = require('assert');
 
 function oneTo5() {
   return Readable.from([1, 2, 3, 4, 5]);
@@ -19,53 +16,45 @@ function oneTo5Async() {
 }
 {
   // Some, find, and every work with a synchronous stream and predicate
-  (async () => {
-    assert.strictEqual(await oneTo5().some((x) => x > 3), true);
-    assert.strictEqual(await oneTo5().every((x) => x > 3), false);
-    assert.strictEqual(await oneTo5().find((x) => x > 3), 4);
-    assert.strictEqual(await oneTo5().some((x) => x > 6), false);
-    assert.strictEqual(await oneTo5().every((x) => x < 6), true);
-    assert.strictEqual(await oneTo5().find((x) => x > 6), undefined);
-    assert.strictEqual(await Readable.from([]).some(() => true), false);
-    assert.strictEqual(await Readable.from([]).every(() => true), true);
-    assert.strictEqual(await Readable.from([]).find(() => true), undefined);
-  })().then(common.mustCall());
+  assert.strictEqual(await oneTo5().some((x) => x > 3), true);
+  assert.strictEqual(await oneTo5().every((x) => x > 3), false);
+  assert.strictEqual(await oneTo5().find((x) => x > 3), 4);
+  assert.strictEqual(await oneTo5().some((x) => x > 6), false);
+  assert.strictEqual(await oneTo5().every((x) => x < 6), true);
+  assert.strictEqual(await oneTo5().find((x) => x > 6), undefined);
+  assert.strictEqual(await Readable.from([]).some(() => true), false);
+  assert.strictEqual(await Readable.from([]).every(() => true), true);
+  assert.strictEqual(await Readable.from([]).find(() => true), undefined);
 }
 
 {
   // Some, find, and every work with an asynchronous stream and synchronous predicate
-  (async () => {
-    assert.strictEqual(await oneTo5Async().some((x) => x > 3), true);
-    assert.strictEqual(await oneTo5Async().every((x) => x > 3), false);
-    assert.strictEqual(await oneTo5Async().find((x) => x > 3), 4);
-    assert.strictEqual(await oneTo5Async().some((x) => x > 6), false);
-    assert.strictEqual(await oneTo5Async().every((x) => x < 6), true);
-    assert.strictEqual(await oneTo5Async().find((x) => x > 6), undefined);
-  })().then(common.mustCall());
+  assert.strictEqual(await oneTo5Async().some((x) => x > 3), true);
+  assert.strictEqual(await oneTo5Async().every((x) => x > 3), false);
+  assert.strictEqual(await oneTo5Async().find((x) => x > 3), 4);
+  assert.strictEqual(await oneTo5Async().some((x) => x > 6), false);
+  assert.strictEqual(await oneTo5Async().every((x) => x < 6), true);
+  assert.strictEqual(await oneTo5Async().find((x) => x > 6), undefined);
 }
 
 {
   // Some, find, and every work on synchronous streams with an asynchronous predicate
-  (async () => {
-    assert.strictEqual(await oneTo5().some(async (x) => x > 3), true);
-    assert.strictEqual(await oneTo5().every(async (x) => x > 3), false);
-    assert.strictEqual(await oneTo5().find(async (x) => x > 3), 4);
-    assert.strictEqual(await oneTo5().some(async (x) => x > 6), false);
-    assert.strictEqual(await oneTo5().every(async (x) => x < 6), true);
-    assert.strictEqual(await oneTo5().find(async (x) => x > 6), undefined);
-  })().then(common.mustCall());
+  assert.strictEqual(await oneTo5().some(async (x) => x > 3), true);
+  assert.strictEqual(await oneTo5().every(async (x) => x > 3), false);
+  assert.strictEqual(await oneTo5().find(async (x) => x > 3), 4);
+  assert.strictEqual(await oneTo5().some(async (x) => x > 6), false);
+  assert.strictEqual(await oneTo5().every(async (x) => x < 6), true);
+  assert.strictEqual(await oneTo5().find(async (x) => x > 6), undefined);
 }
 
 {
   // Some, find, and every work on asynchronous streams with an asynchronous predicate
-  (async () => {
-    assert.strictEqual(await oneTo5Async().some(async (x) => x > 3), true);
-    assert.strictEqual(await oneTo5Async().every(async (x) => x > 3), false);
-    assert.strictEqual(await oneTo5Async().find(async (x) => x > 3), 4);
-    assert.strictEqual(await oneTo5Async().some(async (x) => x > 6), false);
-    assert.strictEqual(await oneTo5Async().every(async (x) => x < 6), true);
-    assert.strictEqual(await oneTo5Async().find(async (x) => x > 6), undefined);
-  })().then(common.mustCall());
+  assert.strictEqual(await oneTo5Async().some(async (x) => x > 3), true);
+  assert.strictEqual(await oneTo5Async().every(async (x) => x > 3), false);
+  assert.strictEqual(await oneTo5Async().find(async (x) => x > 3), 4);
+  assert.strictEqual(await oneTo5Async().some(async (x) => x > 6), false);
+  assert.strictEqual(await oneTo5Async().every(async (x) => x < 6), true);
+  assert.strictEqual(await oneTo5Async().find(async (x) => x > 6), undefined);
 }
 
 {
@@ -74,8 +63,9 @@ function oneTo5Async() {
     await setTimeout();
     assert.strictEqual(stream.destroyed, true);
   }
-  // Some, find, and every short circuit
-  (async () => {
+
+  {
+    // Some, find, and every short circuit
     const someStream = oneTo5();
     await someStream.some(common.mustCall((x) => x > 2, 3));
     await checkDestroyed(someStream);
@@ -92,10 +82,10 @@ function oneTo5Async() {
     await oneTo5().some(common.mustCall(() => false, 5));
     await oneTo5().every(common.mustCall(() => true, 5));
     await oneTo5().find(common.mustCall(() => false, 5));
-  })().then(common.mustCall());
+  }
 
-  // Some, find, and every short circuit async stream/predicate
-  (async () => {
+  {
+    // Some, find, and every short circuit async stream/predicate
     const someStream = oneTo5Async();
     await someStream.some(common.mustCall(async (x) => x > 2, 3));
     await checkDestroyed(someStream);
@@ -112,21 +102,19 @@ function oneTo5Async() {
     await oneTo5Async().some(common.mustCall(async () => false, 5));
     await oneTo5Async().every(common.mustCall(async () => true, 5));
     await oneTo5Async().find(common.mustCall(async () => false, 5));
-  })().then(common.mustCall());
+  }
 }
 
 {
   // Concurrency doesn't affect which value is found.
-  (async () => {
-    const found = await Readable.from([1, 2]).find(async (val) => {
-      if (val === 1) {
-        // eslint-disable-next-line no-restricted-syntax
-        await setTimeout(100);
-      }
-      return true;
-    }, { concurrency: 2 });
-    assert.strictEqual(found, 1);
-  })().then(common.mustCall());
+  const found = await Readable.from([1, 2]).find(async (val) => {
+    if (val === 1) {
+      // eslint-disable-next-line no-restricted-syntax
+      await setTimeout(100);
+    }
+    return true;
+  }, { concurrency: 2 });
+  assert.strictEqual(found, 1);
 }
 
 {
@@ -174,3 +162,13 @@ function oneTo5Async() {
     }, /ERR_INVALID_ARG_TYPE/, `${op} should throw for invalid signal`).then(common.mustCall());
   }
 }
+{
+  for (const op of ['some', 'every', 'find']) {
+    const stream = oneTo5();
+    Object.defineProperty(stream, 'map', {
+      value: common.mustNotCall(() => {}),
+    });
+    // Check that map isn't getting called.
+    stream[op](() => {});
+  }
+}