Skip to content

Commit 0562664

Browse files
committed
fixup! fixup! stream: add iterator helper find
1 parent e5b3ed2 commit 0562664

File tree

2 files changed

+67
-78
lines changed

2 files changed

+67
-78
lines changed

doc/api/stream.md

Lines changed: 21 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1745,7 +1745,8 @@ added: v17.4.0
17451745

17461746
> Stability: 1 - Experimental
17471747
1748-
* `fn` {Function|AsyncFunction} a function to map over every item in the stream.
1748+
* `fn` {Function|AsyncFunction} a function to map over every chunk in the
1749+
stream.
17491750
* `data` {any} a chunk of data from the stream.
17501751
* `options` {Object}
17511752
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1758,16 +1759,16 @@ added: v17.4.0
17581759
* Returns: {Readable} a stream mapped with the function `fn`.
17591760

17601761
This method allows mapping over the stream. The `fn` function will be called
1761-
for every item in the stream. If the `fn` function returns a promise - that
1762+
for every chunk in the stream. If the `fn` function returns a promise - that
17621763
promise will be `await`ed before being passed to the result stream.
17631764

17641765
```mjs
17651766
import { Readable } from 'stream';
17661767
import { Resolver } from 'dns/promises';
17671768

17681769
// With a synchronous mapper.
1769-
for await (const item of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
1770-
console.log(item); // 2, 4, 6, 8
1770+
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
1771+
console.log(chunk); // 2, 4, 6, 8
17711772
}
17721773
// With an asynchronous mapper, making at most 2 queries at a time.
17731774
const resolver = new Resolver();
@@ -1789,7 +1790,7 @@ added: v17.4.0
17891790

17901791
> Stability: 1 - Experimental
17911792
1792-
* `fn` {Function|AsyncFunction} a function to filter items from stream.
1793+
* `fn` {Function|AsyncFunction} a function to filter chunks from the stream.
17931794
* `data` {any} a chunk of data from the stream.
17941795
* `options` {Object}
17951796
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1801,8 +1802,8 @@ added: v17.4.0
18011802
aborted.
18021803
* Returns: {Readable} a stream filtered with the predicate `fn`.
18031804

1804-
This method allows filtering the stream. For each item in the stream the `fn`
1805-
function will be called and if it returns a truthy value, the item will be
1805+
This method allows filtering the stream. For each chunk in the stream the `fn`
1806+
function will be called and if it returns a truthy value, the chunk will be
18061807
passed to the result stream. If the `fn` function returns a promise - that
18071808
promise will be `await`ed.
18081809

@@ -1811,8 +1812,8 @@ import { Readable } from 'stream';
18111812
import { Resolver } from 'dns/promises';
18121813

18131814
// With a synchronous predicate.
1814-
for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1815-
console.log(item); // 3, 4
1815+
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1816+
console.log(chunk); // 3, 4
18161817
}
18171818
// With an asynchronous predicate, making at most 2 queries at a time.
18181819
const resolver = new Resolver();
@@ -1838,7 +1839,7 @@ added: REPLACEME
18381839

18391840
> Stability: 1 - Experimental
18401841
1841-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
1842+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
18421843
* `data` {any} a chunk of data from the stream.
18431844
* `options` {Object}
18441845
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1850,12 +1851,12 @@ added: REPLACEME
18501851
aborted.
18511852
* Returns: {Promise} a promise for when the stream has finished.
18521853

1853-
This method allows iterating a stream. For each item in the stream the
1854+
This method allows iterating a stream. For each chunk in the stream the
18541855
`fn` function will be called. If the `fn` function returns a promise - that
18551856
promise will be `await`ed.
18561857

18571858
This method is different from `for await...of` loops in that it can optionally
1858-
process items concurrently. In addition, a `forEach` iteration can only be
1859+
process chunks concurrently. In addition, a `forEach` iteration can only be
18591860
stopped by having passed a `signal` option and aborting the related
18601861
`AbortController` while `for await...of` can be stopped with `break` or
18611862
`return`. In either case the stream will be destroyed.
@@ -1869,8 +1870,8 @@ import { Readable } from 'stream';
18691870
import { Resolver } from 'dns/promises';
18701871

18711872
// With a synchronous predicate.
1872-
for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1873-
console.log(item); // 3, 4
1873+
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
1874+
console.log(chunk); // 3, 4
18741875
}
18751876
// With an asynchronous predicate, making at most 2 queries at a time.
18761877
const resolver = new Resolver();
@@ -1935,7 +1936,7 @@ added: REPLACEME
19351936

19361937
> Stability: 1 - Experimental
19371938
1938-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
1939+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
19391940
* `data` {any} a chunk of data from the stream.
19401941
* `options` {Object}
19411942
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -1984,7 +1985,7 @@ added: REPLACEME
19841985

19851986
> Stability: 1 - Experimental
19861987
1987-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
1988+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
19881989
* `data` {any} a chunk of data from the stream.
19891990
* `options` {Object}
19901991
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2034,7 +2035,7 @@ added: REPLACEME
20342035

20352036
> Stability: 1 - Experimental
20362037
2037-
* `fn` {Function|AsyncFunction} a function to call on each item of the stream.
2038+
* `fn` {Function|AsyncFunction} a function to call on each chunk of the stream.
20382039
* `data` {any} a chunk of data from the stream.
20392040
* `options` {Object}
20402041
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2084,7 +2085,7 @@ added: REPLACEME
20842085
> Stability: 1 - Experimental
20852086
20862087
* `fn` {Function|AsyncGeneratorFunction|AsyncFunction} a function to map over
2087-
every item in the stream.
2088+
every chunk in the stream.
20882089
* `data` {any} a chunk of data from the stream.
20892090
* `options` {Object}
20902091
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
@@ -2108,8 +2109,8 @@ import { Readable } from 'stream';
21082109
import { createReadStream } from 'fs';
21092110

21102111
// With a synchronous mapper.
2111-
for await (const item of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
2112-
console.log(item); // 1, 1, 2, 2, 3, 3, 4, 4
2112+
for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
2113+
console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
21132114
}
21142115
// With an asynchronous mapper, combine the contents of 4 files
21152116
const concatResult = Readable.from([

test/parallel/test-stream-some-find-every.js renamed to test/parallel/test-stream-some-find-every.mjs

Lines changed: 46 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,8 @@
1-
'use strict';
1+
import * as common from '../common/index.mjs';
2+
import { setTimeout } from 'timers/promises';
3+
import { Readable } from 'stream';
4+
import assert from 'assert';
25

3-
const common = require('../common');
4-
const { setTimeout } = require('timers/promises');
5-
const {
6-
Readable,
7-
} = require('stream');
8-
const assert = require('assert');
96

107
function oneTo5() {
118
return Readable.from([1, 2, 3, 4, 5]);
@@ -19,53 +16,45 @@ function oneTo5Async() {
1916
}
2017
{
2118
// Some, find, and every work with a synchronous stream and predicate
22-
(async () => {
23-
assert.strictEqual(await oneTo5().some((x) => x > 3), true);
24-
assert.strictEqual(await oneTo5().every((x) => x > 3), false);
25-
assert.strictEqual(await oneTo5().find((x) => x > 3), 4);
26-
assert.strictEqual(await oneTo5().some((x) => x > 6), false);
27-
assert.strictEqual(await oneTo5().every((x) => x < 6), true);
28-
assert.strictEqual(await oneTo5().find((x) => x > 6), undefined);
29-
assert.strictEqual(await Readable.from([]).some(() => true), false);
30-
assert.strictEqual(await Readable.from([]).every(() => true), true);
31-
assert.strictEqual(await Readable.from([]).find(() => true), undefined);
32-
})().then(common.mustCall());
19+
assert.strictEqual(await oneTo5().some((x) => x > 3), true);
20+
assert.strictEqual(await oneTo5().every((x) => x > 3), false);
21+
assert.strictEqual(await oneTo5().find((x) => x > 3), 4);
22+
assert.strictEqual(await oneTo5().some((x) => x > 6), false);
23+
assert.strictEqual(await oneTo5().every((x) => x < 6), true);
24+
assert.strictEqual(await oneTo5().find((x) => x > 6), undefined);
25+
assert.strictEqual(await Readable.from([]).some(() => true), false);
26+
assert.strictEqual(await Readable.from([]).every(() => true), true);
27+
assert.strictEqual(await Readable.from([]).find(() => true), undefined);
3328
}
3429

3530
{
3631
// Some, find, and every work with an asynchronous stream and synchronous predicate
37-
(async () => {
38-
assert.strictEqual(await oneTo5Async().some((x) => x > 3), true);
39-
assert.strictEqual(await oneTo5Async().every((x) => x > 3), false);
40-
assert.strictEqual(await oneTo5Async().find((x) => x > 3), 4);
41-
assert.strictEqual(await oneTo5Async().some((x) => x > 6), false);
42-
assert.strictEqual(await oneTo5Async().every((x) => x < 6), true);
43-
assert.strictEqual(await oneTo5Async().find((x) => x > 6), undefined);
44-
})().then(common.mustCall());
32+
assert.strictEqual(await oneTo5Async().some((x) => x > 3), true);
33+
assert.strictEqual(await oneTo5Async().every((x) => x > 3), false);
34+
assert.strictEqual(await oneTo5Async().find((x) => x > 3), 4);
35+
assert.strictEqual(await oneTo5Async().some((x) => x > 6), false);
36+
assert.strictEqual(await oneTo5Async().every((x) => x < 6), true);
37+
assert.strictEqual(await oneTo5Async().find((x) => x > 6), undefined);
4538
}
4639

4740
{
4841
// Some, find, and every work on synchronous streams with an asynchronous predicate
49-
(async () => {
50-
assert.strictEqual(await oneTo5().some(async (x) => x > 3), true);
51-
assert.strictEqual(await oneTo5().every(async (x) => x > 3), false);
52-
assert.strictEqual(await oneTo5().find(async (x) => x > 3), 4);
53-
assert.strictEqual(await oneTo5().some(async (x) => x > 6), false);
54-
assert.strictEqual(await oneTo5().every(async (x) => x < 6), true);
55-
assert.strictEqual(await oneTo5().find(async (x) => x > 6), undefined);
56-
})().then(common.mustCall());
42+
assert.strictEqual(await oneTo5().some(async (x) => x > 3), true);
43+
assert.strictEqual(await oneTo5().every(async (x) => x > 3), false);
44+
assert.strictEqual(await oneTo5().find(async (x) => x > 3), 4);
45+
assert.strictEqual(await oneTo5().some(async (x) => x > 6), false);
46+
assert.strictEqual(await oneTo5().every(async (x) => x < 6), true);
47+
assert.strictEqual(await oneTo5().find(async (x) => x > 6), undefined);
5748
}
5849

5950
{
6051
// Some, find, and every work on asynchronous streams with an asynchronous predicate
61-
(async () => {
62-
assert.strictEqual(await oneTo5Async().some(async (x) => x > 3), true);
63-
assert.strictEqual(await oneTo5Async().every(async (x) => x > 3), false);
64-
assert.strictEqual(await oneTo5Async().find(async (x) => x > 3), 4);
65-
assert.strictEqual(await oneTo5Async().some(async (x) => x > 6), false);
66-
assert.strictEqual(await oneTo5Async().every(async (x) => x < 6), true);
67-
assert.strictEqual(await oneTo5Async().find(async (x) => x > 6), undefined);
68-
})().then(common.mustCall());
52+
assert.strictEqual(await oneTo5Async().some(async (x) => x > 3), true);
53+
assert.strictEqual(await oneTo5Async().every(async (x) => x > 3), false);
54+
assert.strictEqual(await oneTo5Async().find(async (x) => x > 3), 4);
55+
assert.strictEqual(await oneTo5Async().some(async (x) => x > 6), false);
56+
assert.strictEqual(await oneTo5Async().every(async (x) => x < 6), true);
57+
assert.strictEqual(await oneTo5Async().find(async (x) => x > 6), undefined);
6958
}
7059

7160
{
@@ -74,8 +63,9 @@ function oneTo5Async() {
7463
await setTimeout();
7564
assert.strictEqual(stream.destroyed, true);
7665
}
77-
// Some, find, and every short circuit
78-
(async () => {
66+
67+
{
68+
// Some, find, and every short circuit
7969
const someStream = oneTo5();
8070
await someStream.some(common.mustCall((x) => x > 2, 3));
8171
await checkDestroyed(someStream);
@@ -92,10 +82,10 @@ function oneTo5Async() {
9282
await oneTo5().some(common.mustCall(() => false, 5));
9383
await oneTo5().every(common.mustCall(() => true, 5));
9484
await oneTo5().find(common.mustCall(() => false, 5));
95-
})().then(common.mustCall());
85+
}
9686

97-
// Some, find, and every short circuit async stream/predicate
98-
(async () => {
87+
{
88+
// Some, find, and every short circuit async stream/predicate
9989
const someStream = oneTo5Async();
10090
await someStream.some(common.mustCall(async (x) => x > 2, 3));
10191
await checkDestroyed(someStream);
@@ -112,21 +102,19 @@ function oneTo5Async() {
112102
await oneTo5Async().some(common.mustCall(async () => false, 5));
113103
await oneTo5Async().every(common.mustCall(async () => true, 5));
114104
await oneTo5Async().find(common.mustCall(async () => false, 5));
115-
})().then(common.mustCall());
105+
}
116106
}
117107

118108
{
119109
// Concurrency doesn't affect which value is found.
120-
(async () => {
121-
const found = await Readable.from([1, 2]).find(async (val) => {
122-
if (val === 1) {
123-
// eslint-disable-next-line no-restricted-syntax
124-
await setTimeout(100);
125-
}
126-
return true;
127-
}, { concurrency: 2 });
128-
assert.strictEqual(found, 1);
129-
})().then(common.mustCall());
110+
const found = await Readable.from([1, 2]).find(async (val) => {
111+
if (val === 1) {
112+
// eslint-disable-next-line no-restricted-syntax
113+
await setTimeout(100);
114+
}
115+
return true;
116+
}, { concurrency: 2 });
117+
assert.strictEqual(found, 1);
130118
}
131119

132120
{

0 commit comments

Comments
 (0)