From 05626646fe844e312008be566c34735919b6a966 Mon Sep 17 00:00:00 2001 From: Nitzan Uziely Date: Fri, 4 Feb 2022 16:53:07 +0200 Subject: [PATCH] fixup! fixup! stream: add iterator helper find --- doc/api/stream.md | 41 +++---- ...ery.js => test-stream-some-find-every.mjs} | 104 ++++++++---------- 2 files changed, 67 insertions(+), 78 deletions(-) rename test/parallel/{test-stream-some-find-every.js => test-stream-some-find-every.mjs} (57%) diff --git a/doc/api/stream.md b/doc/api/stream.md index 4c5e600d38aabf..747a30a538a5cd 100644 --- a/doc/api/stream.md +++ b/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,7 +1759,7 @@ 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 @@ -1766,8 +1767,8 @@ 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 @@ -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([ diff --git a/test/parallel/test-stream-some-find-every.js b/test/parallel/test-stream-some-find-every.mjs similarity index 57% rename from test/parallel/test-stream-some-find-every.js rename to test/parallel/test-stream-some-find-every.mjs index c63d5c561af31d..975d71b4941281 100644 --- a/test/parallel/test-stream-some-find-every.js +++ b/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); } {