stream: support some and every

This continues on the iterator-helpers work by adding `.some` and `.every` to readable streams. Co-Authored-By: Robert Nagy <[email protected]> PR-URL: #41573 Reviewed-By: Robert Nagy <[email protected]> Reviewed-By: Matteo Collina <[email protected]>

Author: 2 people

doc/api/stream.md

@@ -1864,7 +1864,7 @@ import{Resolver } from 'dns/promises'
 awaitReadable.from([1, 2, 3, 4]).toArray(); // [1, 2, 3, 4]
 
 // Make dns queries concurrently using .map and collect
-// the results into an aray using toArray
+// the results into an array using toArray
 constdnsResults=awaitReadable.from([
 'nodejs.org',
 'openjsf.org',
@@ -1875,6 +1875,104 @@ const dnsResults = await Readable.from([
 },{concurrency:2 }).toArray();
 ```
 
+### `readable.some(fn[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+*`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 evaluating to `true` if `fn` returned a truthy
+ value for at least one of the chunks.
+
+This method is similar to `Array.prototype.some` and calls `fn` on each chunk
+in the stream until the awaited return value is `true` (or any truthy value).
+Once an `fn` call on a chunk awaited return value is truthy, the stream is
+destroyed and the promise is fulfilled with `true`. If none of the `fn`
+calls on the chunks return a truthy value, the promise is fulfilled with
+`false`.
+
+```mjs
+import{Readable } from'stream';
+import{stat } from'fs/promises';
+
+// With a synchronous predicate.
+awaitReadable.from([1, 2, 3, 4]).some((x) => x >2); // true
+awaitReadable.from([1, 2, 3, 4]).some((x) => x <0); // false
+
+// With an asynchronous predicate, making at most 2 file checks at a time.
+constanyBigFile=awaitReadable.from([
+'file1',
+'file2',
+'file3',
+]).some(async (fileName) =>{
+conststats=awaitstat(fileName);
+returnstat.size>1024*1024;
+},{concurrency:2 });
+console.log(anyBigFile); // `true` if any file in the list is bigger than 1MB
+console.log('done'); // Stream has finished
+```
+
+### `readable.every(fn[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+*`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 evaluating to `true` if `fn` returned a truthy
+ value for all of the chunks.
+
+This method is similar to `Array.prototype.every` and calls `fn` on each chunk
+in the stream to check if all awaited return values are truthy value for `fn`.
+Once an `fn` call on a chunk awaited return value is falsy, the stream is
+destroyed and the promise is fulfilled with `false`. If all of the `fn` calls
+on the chunks return a truthy value, the promise is fulfilled with `true`.
+
+```mjs
+import{Readable } from'stream';
+import{stat } from'fs/promises';
+
+// With a synchronous predicate.
+awaitReadable.from([1, 2, 3, 4]).every((x) => x >2); // false
+awaitReadable.from([1, 2, 3, 4]).every((x) => x >0); // true
+
+// With an asynchronous predicate, making at most 2 file checks at a time.
+constallBigFiles=awaitReadable.from([
+'file1',
+'file2',
+'file3',
+]).every(async (fileName) =>{
+conststats=awaitstat(fileName);
+returnstat.size>1024*1024;
+},{concurrency:2 });
+// `true` if all files in the list are bigger than 1MiB
+console.log(allBigFiles);
+console.log('done'); // Stream has finished
+```
+
 ### Duplex and transform streams
 
 #### Class: `stream.Duplex`

lib/internal/streams/operators.js

@@ -10,6 +10,7 @@ const{
  AbortError,
 }=require('internal/errors');
 const{ validateInteger }=require('internal/validators');
+const{ kWeakHandler }=require('internal/event_target');
 
 const{
  ArrayPrototypePush,
@@ -47,6 +48,10 @@ async function * map(fn, options){
 constsignalOpt={ signal };
 
 constabort=()=>ac.abort();
+if(options?.signal?.aborted){
+abort();
+}
+
 options?.signal?.addEventListener('abort',abort);
 
 letnext;
@@ -150,6 +155,40 @@ async function * map(fn, options){
 }
 }
 
+asyncfunctionsome(fn,options){
+// https://tc39.es/proposal-iterator-helpers/#sec-iteratorprototype.some
+// Note that some does short circuit but also closes the iterator if it does
+constac=newAbortController();
+if(options?.signal){
+if(options.signal.aborted){
+ac.abort();
+}
+options.signal.addEventListener('abort',()=>ac.abort(),{
+[kWeakHandler]: this,
+once: true,
+});
+}
+constmapped=this.map(fn,{ ...options,signal: ac.signal});
+forawait(constresultofmapped){
+if(result){
+ac.abort();
+returntrue;
+}
+}
+returnfalse;
+}
+
+asyncfunctionevery(fn,options){
+if(typeoffn!=='function'){
+thrownewERR_INVALID_ARG_TYPE(
+'fn',['Function','AsyncFunction'],fn);
+}
+// https://en.wikipedia.org/wiki/De_Morgan%27s_laws
+return!(awaitsome.call(this,async(x)=>{
+return!(awaitfn(x));
+},options));
+}
+
 asyncfunctionforEach(fn,options){
 if(typeoffn!=='function'){
 thrownewERR_INVALID_ARG_TYPE(
@@ -196,6 +235,8 @@ module.exports.streamReturningOperators ={
 };
 
 module.exports.promiseReturningOperators={
+ every,
  forEach,
  toArray,
+ some,
 };

test/parallel/test-stream-some-every.js

@@ -0,0 +1,95 @@
+'use strict';
+
+constcommon=require('../common');
+const{
+ Readable,
+}=require('stream');
+constassert=require('assert');
+
+functiononeTo5(){
+returnReadable.from([1,2,3,4,5]);
+}
+
+functiononeTo5Async(){
+returnoneTo5().map(async(x)=>{
+awaitPromise.resolve();
+returnx;
+});
+}
+{
+// Some and every work with a synchronous stream and predicate
+(async()=>{
+assert.strictEqual(awaitoneTo5().some((x)=>x>3),true);
+assert.strictEqual(awaitoneTo5().every((x)=>x>3),false);
+assert.strictEqual(awaitoneTo5().some((x)=>x>6),false);
+assert.strictEqual(awaitoneTo5().every((x)=>x<6),true);
+assert.strictEqual(awaitReadable.from([]).some((x)=>true),false);
+assert.strictEqual(awaitReadable.from([]).every((x)=>true),true);
+})().then(common.mustCall());
+}
+
+{
+// Some and every work with an asynchronous stream and synchronous predicate
+(async()=>{
+assert.strictEqual(awaitoneTo5Async().some((x)=>x>3),true);
+assert.strictEqual(awaitoneTo5Async().every((x)=>x>3),false);
+assert.strictEqual(awaitoneTo5Async().some((x)=>x>6),false);
+assert.strictEqual(awaitoneTo5Async().every((x)=>x<6),true);
+})().then(common.mustCall());
+}
+
+{
+// Some and every work on asynchronous streams with an asynchronous predicate
+(async()=>{
+assert.strictEqual(awaitoneTo5().some(async(x)=>x>3),true);
+assert.strictEqual(awaitoneTo5().every(async(x)=>x>3),false);
+assert.strictEqual(awaitoneTo5().some(async(x)=>x>6),false);
+assert.strictEqual(awaitoneTo5().every(async(x)=>x<6),true);
+})().then(common.mustCall());
+}
+
+{
+// Some and every short circuit
+(async()=>{
+awaitoneTo5().some(common.mustCall((x)=>x>2,3));
+awaitoneTo5().every(common.mustCall((x)=>x<3,3));
+// When short circuit isn't possible the whole stream is iterated
+awaitoneTo5().some(common.mustCall((x)=>x>6,5));
+// The stream is destroyed afterwards
+conststream=oneTo5();
+awaitstream.some(common.mustCall((x)=>x>2,3));
+assert.strictEqual(stream.destroyed,true);
+})().then(common.mustCall());
+}
+
+{
+// Support for AbortSignal
+constac=newAbortController();
+assert.rejects(Readable.from([1,2,3]).some(
+()=>newPromise(()=>{}),
+{signal: ac.signal}
+),{
+name: 'AbortError',
+}).then(common.mustCall());
+ac.abort();
+}
+{
+// Support for pre-aborted AbortSignal
+assert.rejects(Readable.from([1,2,3]).some(
+()=>newPromise(()=>{}),
+{signal: AbortSignal.abort()}
+),{
+name: 'AbortError',
+}).then(common.mustCall());
+}
+{
+// Error cases
+assert.rejects(async()=>{
+awaitReadable.from([1]).every(1);
+},/ERR_INVALID_ARG_TYPE/).then(common.mustCall());
+assert.rejects(async()=>{
+awaitReadable.from([1]).every((x)=>x,{
+concurrency: 'Foo'
+});
+},/ERR_OUT_OF_RANGE/).then(common.mustCall());
+}