stream: add drop and take

This adds the `drop` and `take` methods to readable streams allowing users easily drop and take items from the stream. This continues the iterator-helper proposal alignment task. Co-Authored-By: Robert Nagy <[email protected]> PR-URL: #41630 Reviewed-By: Robert Nagy <[email protected]> Reviewed-By: Matteo Collina <[email protected]>

Author: 2 people

doc/api/stream.md

@@ -2020,6 +2020,50 @@ for await (const result of concatResult){
 }
 ```
 
+### `readable.drop(limit[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+*`limit`{number} the number of chunks to drop from the readable.
+*`options`{Object}
+*`signal`{AbortSignal} allows destroying the stream if the signal is
+ aborted.
+* Returns:{Readable} a stream with `limit` chunks dropped.
+
+This method returns a new stream with the first `limit` chunks dropped.
+
+```mjs
+import{Readable } from'stream';
+
+awaitReadable.from([1, 2, 3, 4]).drop(2).toArray(); // [3, 4]
+```
+
+### `readable.take(limit[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+*`limit`{number} the number of chunks to take from the readable.
+*`options`{Object}
+*`signal`{AbortSignal} allows destroying the stream if the signal is
+ aborted.
+* Returns:{Readable} a stream with `limit` chunks taken.
+
+This method returns a new stream with the first `limit` chunks.
+
+```mjs
+import{Readable } from'stream';
+
+awaitReadable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2]
+```
+
 ### Duplex and transform streams
 
 #### Class: `stream.Duplex`

lib/internal/streams/operators.js

@@ -5,6 +5,7 @@ const{AbortController } = require('internal/abort_controller');
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
+ERR_OUT_OF_RANGE,
 },
  AbortError,
 }=require('internal/errors');
@@ -14,6 +15,8 @@ const{kWeakHandler } = require('internal/event_target');
 const{
  ArrayPrototypePush,
  MathFloor,
+ Number,
+ NumberIsNaN,
  Promise,
  PromiseReject,
  PromisePrototypeCatch,
@@ -232,10 +235,62 @@ async function* flatMap(fn, options){
 }
 }
 
+functiontoIntegerOrInfinity(number){
+// We coerce here to align with the spec
+// https://github.com/tc39/proposal-iterator-helpers/issues/169
+number=Number(number);
+if(NumberIsNaN(number)){
+return0;
+}
+if(number<0){
+thrownewERR_OUT_OF_RANGE('number','>= 0',number);
+}
+returnnumber;
+}
+
+functiondrop(number,options){
+number=toIntegerOrInfinity(number);
+returnasyncfunction*drop(){
+if(options?.signal?.aborted){
+thrownewAbortError();
+}
+forawait(constvalofthis){
+if(options?.signal?.aborted){
+thrownewAbortError();
+}
+if(number--<=0){
+yieldval;
+}
+}
+}.call(this);
+}
+
+
+functiontake(number,options){
+number=toIntegerOrInfinity(number);
+returnasyncfunction*take(){
+if(options?.signal?.aborted){
+thrownewAbortError();
+}
+forawait(constvalofthis){
+if(options?.signal?.aborted){
+thrownewAbortError();
+}
+if(number-->0){
+yieldval;
+}else{
+return;
+}
+}
+}.call(this);
+}
+
 module.exports.streamReturningOperators={
+ drop,
  filter,
  flatMap,
  map,
+ take,
 };
 
 module.exports.promiseReturningOperators={

test/parallel/test-stream-drop-take.js

@@ -0,0 +1,96 @@
+'use strict';
+
+constcommon=require('../common');
+const{
+ Readable,
+}=require('stream');
+const{ deepStrictEqual, rejects, throws }=require('assert');
+
+const{ from }=Readable;
+
+constfromAsync=(...args)=>from(...args).map(async(x)=>x);
+
+constnaturals=()=>from(asyncfunction*(){
+leti=1;
+while(true){
+yieldi++;
+}
+}());
+
+{
+// Synchronous streams
+(async()=>{
+deepStrictEqual(awaitfrom([1,2,3]).drop(2).toArray(),[3]);
+deepStrictEqual(awaitfrom([1,2,3]).take(1).toArray(),[1]);
+deepStrictEqual(awaitfrom([]).drop(2).toArray(),[]);
+deepStrictEqual(awaitfrom([]).take(1).toArray(),[]);
+deepStrictEqual(awaitfrom([1,2,3]).drop(1).take(1).toArray(),[2]);
+deepStrictEqual(awaitfrom([1,2]).drop(0).toArray(),[1,2]);
+deepStrictEqual(awaitfrom([1,2]).take(0).toArray(),[]);
+})().then(common.mustCall());
+// Asynchronous streams
+(async()=>{
+deepStrictEqual(awaitfromAsync([1,2,3]).drop(2).toArray(),[3]);
+deepStrictEqual(awaitfromAsync([1,2,3]).take(1).toArray(),[1]);
+deepStrictEqual(awaitfromAsync([]).drop(2).toArray(),[]);
+deepStrictEqual(awaitfromAsync([]).take(1).toArray(),[]);
+deepStrictEqual(awaitfromAsync([1,2,3]).drop(1).take(1).toArray(),[2]);
+deepStrictEqual(awaitfromAsync([1,2]).drop(0).toArray(),[1,2]);
+deepStrictEqual(awaitfromAsync([1,2]).take(0).toArray(),[]);
+})().then(common.mustCall());
+// Infinite streams
+// Asynchronous streams
+(async()=>{
+deepStrictEqual(awaitnaturals().take(1).toArray(),[1]);
+deepStrictEqual(awaitnaturals().drop(1).take(1).toArray(),[2]);
+constnext10=[11,12,13,14,15,16,17,18,19,20];
+deepStrictEqual(awaitnaturals().drop(10).take(10).toArray(),next10);
+deepStrictEqual(awaitnaturals().take(5).take(1).toArray(),[1]);
+})().then(common.mustCall());
+}
+
+{
+// Coercion
+(async()=>{
+// The spec made me do this ^^
+deepStrictEqual(awaitnaturals().take('cat').toArray(),[]);
+deepStrictEqual(awaitnaturals().take('2').toArray(),[1,2]);
+deepStrictEqual(awaitnaturals().take(true).toArray(),[1]);
+})().then(common.mustCall());
+}
+
+{
+// Support for AbortSignal
+constac=newAbortController();
+rejects(
+Readable.from([1,2,3]).take(1,{signal: ac.signal}).toArray(),{
+name: 'AbortError',
+}).then(common.mustCall());
+rejects(
+Readable.from([1,2,3]).drop(1,{signal: ac.signal}).toArray(),{
+name: 'AbortError',
+}).then(common.mustCall());
+ac.abort();
+}
+
+{
+// Support for AbortSignal, already aborted
+constsignal=AbortSignal.abort();
+rejects(
+Readable.from([1,2,3]).take(1,{ signal }).toArray(),{
+name: 'AbortError',
+}).then(common.mustCall());
+}
+
+{
+// Error cases
+constinvalidArgs=[
+-1,
+-Infinity,
+-40,
+];
+
+for(constexampleofinvalidArgs){
+throws(()=>from([]).take(example).toArray(),/ERR_OUT_OF_RANGE/);
+}
+}