* `concurrency` {number} the maximum concurrent invocation of `fn` to call

* `concurrency` {number} the maximum concurrent invocation of `fn` to call

* `fn` {Function|AsyncFunction} a function to call on each item of the stream.

* `data` {any} a chunk of data from the stream.

* `options` {Object}

* `signal` {AbortSignal} aborted if the stream is destroyed allowing to

abort the `fn` call early.

* `options` {Object}

* `concurrency` {number} the maximum concurrent invocation of `fn` to call

on the stream at once. **Default:** `1`.

* `signal` {AbortSignal} allows destroying the stream if the signal is

aborted.

* Returns: {Promise} a promise for when the stream has finished.

This method allows iterating a stream. For each item 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

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.

This method is different from listening to the [`'data'`][] event in that it

uses the [`readable`][] event in the underlying machinary and can limit the

number of concurrent `fn` calls.

import { Readable } from 'stream';

import { Resolver } from 'dns/promises';

for await (const item of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {

console.log(item); // 3, 4

}

const resolver = new Resolver();

const dnsResults = await Readable.from([

'nodejs.org',

'openjsf.org',

'www.linuxfoundation.org',

]).map(async (domain) => {

const { address } = await resolver.resolve4(domain, { ttl: true });

return address;

}, { concurrency: 2 });

await dnsResults.forEach((result) => {

// Logs result, similar to `for await (const result of dnsResults)`

console.log(result);

});

console.log('done'); // Stream has finished

'fn', ['Function', 'AsyncFunction'], fn);

async function forEach(fn, options) {

if (typeof fn !== 'function') {

throw new ERR_INVALID_ARG_TYPE(

'fn', ['Function', 'AsyncFunction'], fn);

}

async function forEachFn(value, options) {

await fn(value, options);

return kEmpty;

}

// eslint-disable-next-line no-unused-vars

for await (const unused of this.map(forEachFn, options));

throw new ERR_INVALID_ARG_TYPE(

'fn', ['Function', 'AsyncFunction'], fn);

module.exports.streamReturningOperators = {

filter,

module.exports.promiseReturningOperators = {

forEach,

const {

streamReturningOperators,

promiseReturningOperators,

} = require('internal/streams/operators');

for (const key of ObjectKeys(streamReturningOperators)) {

const op = streamReturningOperators[key];

for (const key of ObjectKeys(promiseReturningOperators)) {

const op = promiseReturningOperators[key];

Stream.Readable.prototype[key] = function(...args) {

return ReflectApply(op, this, args);

};

const common = require('../common');

const {

Readable,

} = require('stream');

const assert = require('assert');

const { setTimeout } = require('timers/promises');

// forEach works on synchronous streams with a synchronous predicate

const stream = Readable.from([1, 2, 3]);

const result = [1, 2, 3];

(async () => {

await stream.forEach((value) => assert.strictEqual(value, result.shift()));

})().then(common.mustCall());

// forEach works an asynchronous streams

const stream = Readable.from([1, 2, 3]).filter(async (x) => {

await Promise.resolve();

return true;

});

const result = [1, 2, 3];

(async () => {

await stream.forEach((value) => assert.strictEqual(value, result.shift()));

})().then(common.mustCall());

// forEach works on asynchronous streams with a asynchronous forEach fn

const stream = Readable.from([1, 2, 3]).filter(async (x) => {

await Promise.resolve();

return true;

});

const result = [1, 2, 3];

(async () => {

await stream.forEach(async (value) => {

await Promise.resolve();

assert.strictEqual(value, result.shift());

});

})().then(common.mustCall());

// Concurrency + AbortSignal

const ac = new AbortController();

let calls = 0;

const forEachPromise =

Readable.from([1, 2, 3, 4]).forEach(async (_, { signal }) => {

calls++;

await setTimeout(100, { signal });

}, { signal: ac.signal, concurrency: 2 });

// pump

assert.rejects(async () => {

await forEachPromise;

}, {

name: 'AbortError',

}).then(common.mustCall());

setImmediate(() => {

ac.abort();

assert.strictEqual(calls, 2);

});

// Error cases

assert.rejects(async () => {

await Readable.from([1]).forEach(1);

}, /ERR_INVALID_ARG_TYPE/).then(common.mustCall());

assert.rejects(async () => {

await Readable.from([1]).forEach((x) => x, {

concurrency: 'Foo'

});

}, /ERR_OUT_OF_RANGE/).then(common.mustCall());

assert.rejects(async () => {

await Readable.from([1]).forEach((x) => x, 1);

}, /ERR_INVALID_ARG_TYPE/).then(common.mustCall());

// Test result is a Promise

const stream = Readable.from([1, 2, 3, 4, 5]).forEach((_) => true);

assert.strictEqual(typeof stream.then, 'function');