FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: add iterator helper find · nodejs/node@1ae6485 · GitHub

/ node Public

Commit 1ae6485

Browse files
Linkgoron
authored and
ruyadorno
committed
stream: add iterator helper find
Continue iterator-helpers work by adding `find` to readable streams. PR-URL: #41849 Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent d641eb7 commit 1ae6485

7 files changed

Lines changed: 282 additions & 153 deletions

File tree

‎doc/api/stream.md‎

Lines changed: 70 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1691,7 +1691,8 @@ added: v17.4.0
16911691

16921692
> Stability: 1 - Experimental
16931693
1694-
* `fn` {Function|AsyncFunction} a function to map over every item in the stream.
1694+
* `fn` {Function|AsyncFunction} a function to map over every chunk in the
1695+
stream.
16951696
* `data` {any} a chunk of data from the stream.
16961697
* `options` {Object}
16971698
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1704,16 +1705,16 @@ added: v17.4.0
17041705
* Returns: {Readable} a stream mapped with the function `fn`.
17051706

17061707
This method allows mapping over the stream. The `fn` function will be called
1707-
for every item in the stream. If the `fn` function returns a promise - that
1708+
for every chunk in the stream. If the `fn` function returns a promise - that
17081709
promise will be `await`ed before being passed to the result stream.
17091710

17101711
```mjs
17111712
import { Readable } from 'stream';
17121713
import { Resolver } from 'dns/promises';
17131714

17141715
// With a synchronous mapper.
1715-
for await (const item of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
1716-
console.log(item); // 2, 4, 6, 8
1716+
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
1717+
console.log(chunk); // 2, 4, 6, 8
17171718
}
17181719
// With an asynchronous mapper, making at most 2 queries at a time.
17191720
const resolver = new Resolver();
@@ -1735,7 +1736,7 @@ added: v17.4.0
17351736

17361737
> Stability: 1 - Experimental
17371738
1738-
* `fn` {Function|AsyncFunction} a function to filter items from stream.
1739+
* `fn` {Function|AsyncFunction} a function to filter chunks from the stream.
17391740
* `data` {any} a chunk of data from the stream.
17401741
* `options` {Object}
17411742
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1747,8 +1748,8 @@ added: v17.4.0
17471748
aborted.
17481749
* Returns: {Readable} a stream filtered with the predicate `fn`.
17491750

1750-
This method allows filtering the stream. For each item in the stream the `fn`
1751-
function will be called and if it returns a truthy value, the item will be
1751+
This method allows filtering the stream. For each chunk in the stream the `fn`
1752+
function will be called and if it returns a truthy value, the chunk will be
17521753
passed to the result stream. If the `fn` function returns a promise - that
17531754
promise will be `await`ed.
17541755

@@ -1757,8 +1758,8 @@ import { Readable } from 'stream';
17571758
import { Resolver } from 'dns/promises';
17581759

17591760
// With a synchronous predicate.
1760-
for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1761-
console.log(item); // 3, 4
1761+
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1762+
console.log(chunk); // 3, 4
17621763
}
17631764
// With an asynchronous predicate, making at most 2 queries at a time.
17641765
const resolver = new Resolver();
@@ -1784,7 +1785,7 @@ added: REPLACEME
17841785

17851786
> Stability: 1 - Experimental
17861787
1787-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
1788+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
17881789
* `data` {any} a chunk of data from the stream.
17891790
* `options` {Object}
17901791
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1796,12 +1797,12 @@ added: REPLACEME
17961797
aborted.
17971798
* Returns: {Promise} a promise for when the stream has finished.
17981799

1799-
This method allows iterating a stream. For each item in the stream the
1800+
This method allows iterating a stream. For each chunk in the stream the
18001801
`fn` function will be called. If the `fn` function returns a promise - that
18011802
promise will be `await`ed.
18021803

18031804
This method is different from `for await...of` loops in that it can optionally
1804-
process items concurrently. In addition, a `forEach` iteration can only be
1805+
process chunks concurrently. In addition, a `forEach` iteration can only be
18051806
stopped by having passed a `signal` option and aborting the related
18061807
`AbortController` while `for await...of` can be stopped with `break` or
18071808
`return`. In either case the stream will be destroyed.
@@ -1815,8 +1816,8 @@ import { Readable } from 'stream';
18151816
import { Resolver } from 'dns/promises';
18161817

18171818
// With a synchronous predicate.
1818-
for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1819-
console.log(item); // 3, 4
1819+
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1820+
console.log(chunk); // 3, 4
18201821
}
18211822
// With an asynchronous predicate, making at most 2 queries at a time.
18221823
const resolver = new Resolver();
@@ -1881,7 +1882,7 @@ added: REPLACEME
18811882

18821883
> Stability: 1 - Experimental
18831884
1884-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
1885+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
18851886
* `data` {any} a chunk of data from the stream.
18861887
* `options` {Object}
18871888
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1922,6 +1923,56 @@ console.log(anyBigFile); // `true` if any file in the list is bigger than 1MB
19221923
console.log('done'); // Stream has finished
19231924
```
19241925

1926+
### `readable.find(fn[, options])`
1927+
1928+
<!-- YAML
1929+
added: REPLACEME
1930+
-->
1931+
1932+
> Stability: 1 - Experimental
1933+
1934+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
1935+
* `data` {any} a chunk of data from the stream.
1936+
* `options` {Object}
1937+
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
1938+
abort the `fn` call early.
1939+
* `options` {Object}
1940+
* `concurrency` {number} the maximum concurrent invocation of `fn` to call
1941+
on the stream at once. **Default:** `1`.
1942+
* `signal` {AbortSignal} allows destroying the stream if the signal is
1943+
aborted.
1944+
* Returns: {Promise} a promise evaluating to the first chunk for which `fn`
1945+
evaluated with a truthy value, or `undefined` if no element was found.
1946+
1947+
This method is similar to `Array.prototype.find` and calls `fn` on each chunk
1948+
in the stream to find a chunk with a truthy value for `fn`. Once an `fn` call's
1949+
awaited return value is truthy, the stream is destroyed and the promise is
1950+
fulfilled with value for which `fn` returned a truthy value. If all of the
1951+
`fn` calls on the chunks return a falsy value, the promise is fulfilled with
1952+
`undefined`.
1953+
1954+
```mjs
1955+
import { Readable } from 'stream';
1956+
import { stat } from 'fs/promises';
1957+
1958+
// With a synchronous predicate.
1959+
await Readable.from([1, 2, 3, 4]).find((x) => x > 2); // 3
1960+
await Readable.from([1, 2, 3, 4]).find((x) => x > 0); // 1
1961+
await Readable.from([1, 2, 3, 4]).find((x) => x > 10); // undefined
1962+
1963+
// With an asynchronous predicate, making at most 2 file checks at a time.
1964+
const foundBigFile = await Readable.from([
1965+
'file1',
1966+
'file2',
1967+
'file3',
1968+
]).find(async (fileName) => {
1969+
const stats = await stat(fileName);
1970+
return stat.size > 1024 * 1024;
1971+
}, { concurrency: 2 });
1972+
console.log(foundBigFile); // File name of large file, if any file in the list is bigger than 1MB
1973+
console.log('done'); // Stream has finished
1974+
```
1975+
19251976
### `readable.every(fn[, options])`
19261977

19271978
<!-- YAML
@@ -1930,7 +1981,7 @@ added: REPLACEME
19301981

19311982
> Stability: 1 - Experimental
19321983
1933-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
1984+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
19341985
* `data` {any} a chunk of data from the stream.
19351986
* `options` {Object}
19361987
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1980,7 +2031,7 @@ added: REPLACEME
19802031
> Stability: 1 - Experimental
19812032
19822033
* `fn` {Function|AsyncGeneratorFunction|AsyncFunction} a function to map over
1983-
every item in the stream.
2034+
every chunk in the stream.
19842035
* `data` {any} a chunk of data from the stream.
19852036
* `options` {Object}
19862037
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2004,8 +2055,8 @@ import { Readable } from 'stream';
20042055
import { createReadStream } from 'fs';
20052056

20062057
// With a synchronous mapper.
2007-
for await (const item of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
2008-
console.log(item); // 1, 1, 2, 2, 3, 3, 4, 4
2058+
for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
2059+
console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
20092060
}
20102061
// With an asynchronous mapper, combine the contents of 4 files
20112062
const concatResult = Readable.from([

‎lib/internal/streams/operators.js‎

Lines changed: 14 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -186,31 +186,9 @@ function asIndexedPairs(options = undefined) {
186186
}
187187

188188
async function some(fn, options) {
189-
if (options != null && typeof options !== 'object') {
190-
throw new ERR_INVALID_ARG_TYPE('options', ['Object']);
191-
}
192-
if (options?.signal != null) {
193-
validateAbortSignal(options.signal, 'options.signal');
194-
}
195-
196-
// https://tc39.es/proposal-iterator-helpers/#sec-iteratorprototype.some
197-
// Note that some does short circuit but also closes the iterator if it does
198-
const ac = new AbortController();
199-
if (options?.signal) {
200-
if (options.signal.aborted) {
201-
ac.abort();
202-
}
203-
options.signal.addEventListener('abort', () => ac.abort(), {
204-
[kWeakHandler]: this,
205-
once: true,
206-
});
207-
}
208-
const mapped = this.map(fn, { ...options, signal: ac.signal });
209-
for await (const result of mapped) {
210-
if (result) {
211-
ac.abort();
212-
return true;
213-
}
189+
// eslint-disable-next-line no-unused-vars
190+
for await (const unused of filter.call(this, fn, options)) {
191+
return true;
214192
}
215193
return false;
216194
}
@@ -226,6 +204,13 @@ async function every(fn, options) {
226204
}, options));
227205
}
228206

207+
async function find(fn, options) {
208+
for await (const result of filter.call(this, fn, options)) {
209+
return result;
210+
}
211+
return undefined;
212+
}
213+
229214
async function forEach(fn, options) {
230215
if (typeof fn !== 'function') {
231216
throw new ERR_INVALID_ARG_TYPE(
@@ -236,7 +221,7 @@ async function forEach(fn, options) {
236221
return kEmpty;
237222
}
238223
// eslint-disable-next-line no-unused-vars
239-
for await (const unused of this.map(forEachFn, options));
224+
for await (const unused of map.call(this, forEachFn, options));
240225
}
241226

242227
function filter(fn, options) {
@@ -250,7 +235,7 @@ function filter(fn, options) {
250235
}
251236
return kEmpty;
252237
}
253-
return this.map(filterFn, options);
238+
return map.call(this, filterFn, options);
254239
}
255240

256241
// Specific to provide better error to reduce since the argument is only
@@ -329,7 +314,7 @@ async function toArray(options) {
329314
}
330315

331316
function flatMap(fn, options) {
332-
const values = this.map(fn, options);
317+
const values = map.call(this, fn, options);
333318
return async function* flatMap() {
334319
for await (const val of values) {
335320
yield* val;
@@ -415,4 +400,5 @@ module.exports.promiseReturningOperators = {
415400
reduce,
416401
toArray,
417402
some,
403+
find,
418404
};

‎test/parallel/test-stream-filter.js‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,3 +98,11 @@ const { setTimeout } = require('timers/promises');
9898
const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => true);
9999
assert.strictEqual(stream.readable, true);
100100
}
101+
{
102+
const stream = Readable.from([1, 2, 3, 4, 5]);
103+
Object.defineProperty(stream, 'map', {
104+
value: common.mustNotCall(() => {}),
105+
});
106+
// Check that map isn't getting called.
107+
stream.filter(() => true);
108+
}

‎test/parallel/test-stream-flatMap.js‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,3 +121,11 @@ function oneTo5() {
121121
const stream = oneTo5().flatMap((x) => x);
122122
assert.strictEqual(stream.readable, true);
123123
}
124+
{
125+
const stream = oneTo5();
126+
Object.defineProperty(stream, 'map', {
127+
value: common.mustNotCall(() => {}),
128+
});
129+
// Check that map isn't getting called.
130+
stream.flatMap(() => true);
131+
}

‎test/parallel/test-stream-forEach.js‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,3 +84,11 @@ const { setTimeout } = require('timers/promises');
8484
const stream = Readable.from([1, 2, 3, 4, 5]).forEach((_) => true);
8585
assert.strictEqual(typeof stream.then, 'function');
8686
}
87+
{
88+
const stream = Readable.from([1, 2, 3, 4, 5]);
89+
Object.defineProperty(stream, 'map', {
90+
value: common.mustNotCall(() => {}),
91+
});
92+
// Check that map isn't getting called.
93+
stream.forEach(() => true);
94+
}

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL