Skip to content

Commit 013df4b

Browse files
committed
core review + tests
1 parent f14286f commit 013df4b

File tree

4 files changed

+31
-19
lines changed

4 files changed

+31
-19
lines changed

doc/api/stream.md

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1751,7 +1751,7 @@ added: REPLACEME
17511751
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
17521752
abort the `fn` call early.
17531753
* `options` {Object}
1754-
* `concurrency` {number} the maximal concurrent invocation of `fn` to call
1754+
* `concurrency` {number} the maximum concurrent invocation of `fn` to call
17551755
on the stream at once. **Default:** `1`.
17561756
* `signal` {AbortSignal} allows destroying the stream if the signal is
17571757
aborted.
@@ -1795,7 +1795,7 @@ added: REPLACEME
17951795
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
17961796
abort the `fn` call early.
17971797
* `options` {Object}
1798-
* `concurrency` {number} the maximal concurrent invocation of `fn` to call
1798+
* `concurrency` {number} the maximum concurrent invocation of `fn` to call
17991799
on the stream at once. **Default:** `1`.
18001800
* `signal` {AbortSignal} allows destroying the stream if the signal is
18011801
aborted.
@@ -1844,7 +1844,7 @@ added: REPLACEME
18441844
* `signal` {AbortSignal} aborted if the stream is destroyed allowing to
18451845
abort the `fn` call early.
18461846
* `options` {Object}
1847-
* `concurrency` {number} the maximal concurrent invocation of `fn` to call
1847+
* `concurrency` {number} the maximum concurrent invocation of `fn` to call
18481848
on the stream at once. **Default:** `1`.
18491849
* `signal` {AbortSignal} allows destroying the stream if the signal is
18501850
aborted.
@@ -1854,11 +1854,11 @@ This method allows iterating a stream. For each item in the stream the
18541854
`fn` function will be called. If the `fn` function returns a promise - that
18551855
promise will be `await`ed.
18561856

1857-
This method is different from `for... await` loops in that it supports setting
1858-
the maximal concurrent invocation of `fn` through the `concurrency` option. It
1859-
is also possible to `break` from a `for... await` destroying the stream but
1860-
`forEach` is only breakable by passing it a `signal` and aborting the related
1861-
`AbortController`.
1857+
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+
stopped by having passed a `signal` option and aborting the related
1860+
`AbortController` while `for await...of` can be stopped with `break` or
1861+
`return`. In either case the stream will be destroyed.
18621862

18631863
This method is different from listening to the [`'data'`][] event in that it
18641864
uses the [`readable`][] event in the underlying machinary and can limit the

lib/internal/streams/operators.js

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ const kEof = Symbol('kEof');
2323
async function * map(fn, options) {
2424
if (typeof fn !== 'function') {
2525
throw new ERR_INVALID_ARG_TYPE(
26-
'fn', ['Function', 'AsyncFunction'], this);
26+
'fn', ['Function', 'AsyncFunction'], fn);
2727
}
2828

2929
if (options != null && typeof options !== 'object') {
@@ -150,7 +150,7 @@ async function * map(fn, options) {
150150
async function forEach(fn, options) {
151151
if (typeof fn !== 'function') {
152152
throw new ERR_INVALID_ARG_TYPE(
153-
'fn', ['Function', 'AsyncFunction'], this);
153+
'fn', ['Function', 'AsyncFunction'], fn);
154154
}
155155
async function forEachFn(value, options) {
156156
await fn(value, options);
@@ -163,7 +163,7 @@ async function forEach(fn, options) {
163163
async function * filter(fn, options) {
164164
if (typeof fn !== 'function') {
165165
throw new ERR_INVALID_ARG_TYPE(
166-
'fn', ['Function', 'AsyncFunction'], this);
166+
'fn', ['Function', 'AsyncFunction'], fn);
167167
}
168168
async function filterFn(value, options) {
169169
if (await fn(value, options)) {
@@ -174,8 +174,11 @@ async function * filter(fn, options) {
174174
yield* this.map(filterFn, options);
175175
}
176176

177-
module.exports = {
177+
module.exports.streamReturningOperators = {
178178
filter,
179-
forEach,
180179
map,
181180
};
181+
182+
module.exports.promiseReturningOperators = {
183+
forEach,
184+
};

lib/stream.js

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,10 @@ const {
3131
promisify: { custom: customPromisify },
3232
} = require('internal/util');
3333

34-
const operators = require('internal/streams/operators');
34+
const {
35+
streamReturningOperators,
36+
promiseReturningOperators,
37+
} = require('internal/streams/operators');
3538
const compose = require('internal/streams/compose');
3639
const { pipeline } = require('internal/streams/pipeline');
3740
const { destroyer } = require('internal/streams/destroy');
@@ -46,12 +49,18 @@ Stream.isDisturbed = utils.isDisturbed;
4649
Stream.isErrored = utils.isErrored;
4750
Stream.isReadable = utils.isReadable;
4851
Stream.Readable = require('internal/streams/readable');
49-
for (const key of ObjectKeys(operators)) {
50-
const op = operators[key];
52+
for (const key of ObjectKeys(streamReturningOperators)) {
53+
const op = streamReturningOperators[key];
5154
Stream.Readable.prototype[key] = function(...args) {
5255
return Stream.Readable.from(ReflectApply(op, this, args));
5356
};
5457
}
58+
for (const key of ObjectKeys(promiseReturningOperators)) {
59+
const op = promiseReturningOperators[key];
60+
Stream.Readable.prototype[key] = function(...args) {
61+
return ReflectApply(op, this, args);
62+
};
63+
}
5564
Stream.Writable = require('internal/streams/writable');
5665
Stream.Duplex = require('internal/streams/duplex');
5766
Stream.Transform = require('internal/streams/transform');

test/parallel/test-stream-forEach.js

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -68,15 +68,15 @@ const { setTimeout } = require('timers/promises');
6868
{
6969
// Error cases
7070
assert.rejects(async () => {
71-
Readable.from([1]).forEach(1);
71+
await Readable.from([1]).forEach(1);
7272
}, /ERR_INVALID_ARG_TYPE/).then(common.mustCall());
7373
assert.rejects(async () => {
74-
Readable.from([1]).forEach((x) => x, {
74+
await Readable.from([1]).forEach((x) => x, {
7575
concurrency: 'Foo'
7676
});
7777
}, /ERR_OUT_OF_RANGE/).then(common.mustCall());
7878
assert.rejects(async () => {
79-
Readable.from([1]).forEach((x) => x, 1);
79+
await Readable.from([1]).forEach((x) => x, 1);
8080
}, /ERR_INVALID_ARG_TYPE/).then(common.mustCall());
8181
}
8282
{

0 commit comments

Comments
 (0)