stream: add filter method to readable

This continues the work in #40815 to
make streams compatible with upcoming ECMAScript language features. It
adds an experimental `filter` api to streams and tests/docs for it.
See https://github.com/tc39/proposal-iterator-helpers/
Co-Authored-By: Robert Nagy <ronagy@icloud.com>
PR-URL: #41354
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: benjamingr

doc/api/stream.md

@@ -1726,6 +1726,55 @@ for await (const result of dnsResults) {
 }
 ```
 
+### `readable.filter(fn[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+*`fn` {Function|AsyncFunction} a function to filter items from 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 maximal 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: {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
+passed to the result stream. If the `fn` function returns a promise - that
+promise will be `await`ed.
+
+```mjs
+import { Readable } from'stream';
+import { Resolver } from'dns/promises';
+
+// With a synchronous predicate.
+forawait (constitemofReadable.from([1, 2, 3, 4]).filter((x) => x >2)) {
+console.log(item); // 3, 4
+}
+// With an asynchronous predicate, making at most 2 queries at a time.
+constresolver=newResolver();
+constdnsResults=awaitReadable.from([
+'nodejs.org',
+'openjsf.org',
+'www.linuxfoundation.org',
+]).filter(async (domain) => {
+const { address } =awaitresolver.resolve4(domain, { ttl:true });
+returnaddress.ttl>60;
+}, { concurrency:2 });
+forawait (constresultofdnsResults) {
+// Logs domains with more than 60 seconds on the resolved dns record.
+console.log(result);
+}
+```
+
 ### Duplex and transform streams
 
 #### Class: `stream.Duplex`

lib/internal/streams/operators.js

@@ -147,6 +147,20 @@ async function * map(fn, options) {
 }
 }
 
+asyncfunction*filter(fn,options){
+if(typeoffn!=='function'){
+throw(newERR_INVALID_ARG_TYPE(
+'fn',['Function','AsyncFunction'],this));
+}
+asyncfunctionfilterFn(value,options){
+if(awaitfn(value,options)){
+returnvalue;
+}
+returnkEmpty;
+}
+yield*this.map(filterFn,options);
+}
 module.exports={
  map,
+ filter
 };

test/parallel/test-stream-filter.js

@@ -0,0 +1,109 @@
+'use strict';
+
+constcommon=require('../common');
+const{
+ Readable,
+}=require('stream');
+constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
+
+{
+// Filter works on synchronous streams with a synchronous predicate
+conststream=Readable.from([1,2,3,4,5]).filter((x)=>x<3);
+constresult=[1,2];
+(async()=>{
+forawait(constitemofstream){
+assert.strictEqual(item,result.shift());
+}
+})().then(common.mustCall());
+}
+
+{
+// Filter works on synchronous streams with an asynchronous predicate
+conststream=Readable.from([1,2,3,4,5]).filter(async(x)=>{
+awaitPromise.resolve();
+returnx>3;
+});
+constresult=[4,5];
+(async()=>{
+forawait(constitemofstream){
+assert.strictEqual(item,result.shift());
+}
+})().then(common.mustCall());
+}
+
+{
+// Map works on asynchronous streams with a asynchronous mapper
+conststream=Readable.from([1,2,3,4,5]).map(async(x)=>{
+awaitPromise.resolve();
+returnx+x;
+}).filter((x)=>x>5);
+constresult=[6,8,10];
+(async()=>{
+forawait(constitemofstream){
+assert.strictEqual(item,result.shift());
+}
+})().then(common.mustCall());
+}
+
+{
+// Concurrency + AbortSignal
+constac=newAbortController();
+letcalls=0;
+conststream=Readable.from([1,2,3,4]).filter(async(_,{ signal })=>{
+calls++;
+awaitsetTimeout(100,{ signal });
+},{signal: ac.signal,concurrency: 2});
+// pump
+assert.rejects(async()=>{
+forawait(constitemofstream){
+// nope
+console.log(item);
+}
+},{
+name: 'AbortError',
+}).then(common.mustCall());
+
+setImmediate(()=>{
+ac.abort();
+assert.strictEqual(calls,2);
+});
+}
+
+{
+// Concurrency result order
+conststream=Readable.from([1,2]).filter(async(item,{ signal })=>{
+awaitsetTimeout(10-item,{ signal });
+returntrue;
+},{concurrency: 2});
+
+(async()=>{
+constexpected=[1,2];
+forawait(constitemofstream){
+assert.strictEqual(item,expected.shift());
+}
+})().then(common.mustCall());
+}
+
+{
+// Error cases
+assert.rejects(async()=>{
+// eslint-disable-next-line no-unused-vars
+forawait(constunusedofReadable.from([1]).filter(1));
+},/ERR_INVALID_ARG_TYPE/).then(common.mustCall());
+assert.rejects(async()=>{
+// eslint-disable-next-line no-unused-vars
+forawait(const_ofReadable.from([1]).filter((x)=>x,{
+concurrency: 'Foo'
+}));
+},/ERR_OUT_OF_RANGE/).then(common.mustCall());
+assert.rejects(async()=>{
+// eslint-disable-next-line no-unused-vars
+forawait(const_ofReadable.from([1]).filter((x)=>x,1));
+},/ERR_INVALID_ARG_TYPE/).then(common.mustCall());
+}
+{
+// Test result is a Readable
+conststream=Readable.from([1,2,3,4,5]).filter((x)=>true);
+assert.strictEqual(stream.readable,true);
+}