stream: support abort signal

PR-URL: #36061 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Robert Nagy <[email protected]>

Author: benjamingr

doc/api/stream.md

@@ -45,8 +45,8 @@ There are four fundamental stream types within Node.js:
  is written and read (for example, [`zlib.createDeflate()`][]).
 
 Additionally, this module includes the utility functions
-[`stream.pipeline()`][], [`stream.finished()`][] and
-[`stream.Readable.from()`][].
+[`stream.pipeline()`][], [`stream.finished()`][], [`stream.Readable.from()`][]
+and [`stream.addAbortSignal()`][].
 
 ### Streams Promises API
 <!-- YAML
@@ -1799,6 +1799,55 @@ Calling `Readable.from(string)` or `Readable.from(buffer)` will not have
 the strings or buffers be iterated to match the other streams semantics
 for performance reasons.
 
+### `stream.addAbortSignal(signal, stream)`
+<!-- YAML
+added: REPLACEME
+-->
+*`signal`{AbortSignal} A signal representing possible cancellation
+*`stream`{Stream} a stream to attach a signal to
+
+Attaches an AbortSignal to a readable or writeable stream. This lets code
+control stream destruction using an `AbortController`.
+
+Calling `abort` on the `AbortController` corresponding to the passed
+`AbortSignal` will behave the same way as calling `.destroy(new AbortError())`
+on the stream.
+
+```js
+constfs=require('fs');
+
+constcontroller=newAbortController();
+constread=addAbortSignal(
+controller.signal,
+fs.createReadStream(('object.json'))
+);
+// Later, abort the operation closing the stream
+controller.abort();
+```
+
+Or using an `AbortSignal` with a readable stream as an async iterable:
+
+```js
+constcontroller=newAbortController();
+setTimeout(() =>controller.abort(), 10_000); // set a timeout
+conststream=addAbortSignal(
+controller.signal,
+fs.createReadStream(('object.json'))
+);
+(async () =>{
+try{
+forawait (constchunkofstream){
+awaitprocess(chunk);
+ }
+ } catch (e){
+if (e.name==='AbortError'){
+// The operation was cancelled
+ } else{
+throw e;
+ }
+ }
+})();
+```
 ## API for stream implementers
 
 <!--type=misc-->
@@ -3123,6 +3172,7 @@ contain multi-byte characters.
 [`stream.finished()`]: #stream_stream_finished_stream_options_callback
 [`stream.pipe()`]: #stream_readable_pipe_destination_options
 [`stream.pipeline()`]: #stream_stream_pipeline_source_transforms_destination_callback
+[`stream.addAbortSignal()`]: #stream_stream_addabortsignal_signal_stream
 [`stream.uncork()`]: #stream_writable_uncork
 [`stream.unpipe()`]: #stream_readable_unpipe_destination
 [`stream.wrap()`]: #stream_readable_wrap_stream

lib/_http_client.js

@@ -51,7 +51,7 @@ const{Buffer } = require('buffer');
 const{ defaultTriggerAsyncIdScope }=require('internal/async_hooks');
 const{URL, urlToOptions, searchParamsSymbol }=require('internal/url');
 const{ kOutHeaders, kNeedDrain }=require('internal/http');
-const{AbortError,connResetException, codes }=require('internal/errors');
+const{ connResetException, codes }=require('internal/errors');
 const{
 ERR_HTTP_HEADERS_SENT,
 ERR_INVALID_ARG_TYPE,
@@ -61,14 +61,15 @@ const{
 }=codes;
 const{
  validateInteger,
- validateAbortSignal,
 }=require('internal/validators');
 const{ getTimerDuration }=require('internal/timers');
 const{
 DTRACE_HTTP_CLIENT_REQUEST,
 DTRACE_HTTP_CLIENT_RESPONSE
 }=require('internal/dtrace');
 
+const{ addAbortSignal }=require('stream');
+
 constINVALID_PATH_REGEX=/[^\u0021-\u00ff]/;
 constkError=Symbol('kError');
 
@@ -174,12 +175,7 @@ function ClientRequest(input, options, cb){
 
 constsignal=options.signal;
 if(signal){
-validateAbortSignal(signal,'options.signal');
-constlistener=(e)=>this.destroy(newAbortError());
-signal.addEventListener('abort',listener);
-this.once('close',()=>{
-signal.removeEventListener('abort',listener);
-});
+addAbortSignal(signal,this);
 }
 letmethod=options.method;
 constmethodIsString=(typeofmethod==='string');

lib/internal/streams/add-abort-signal.js

@@ -0,0 +1,41 @@
+'use strict';
+
+const{
+ AbortError,
+ codes,
+}=require('internal/errors');
+
+consteos=require('internal/streams/end-of-stream');
+const{ERR_INVALID_ARG_TYPE}=codes;
+
+// This method is inlined here for readable-stream
+// https://github.com/nodejs/node/pull/36061#discussion_r533718029
+constvalidateAbortSignal=(signal,name)=>{
+if(signal!==undefined&&
+(signal===null||
+typeofsignal!=='object'||
+!('aborted'insignal))){
+thrownewERR_INVALID_ARG_TYPE(name,'AbortSignal',signal);
+}
+};
+
+functionisStream(obj){
+return!!(obj&&typeofobj.pipe==='function');
+}
+
+module.exports=functionaddAbortSignal(signal,stream){
+validateAbortSignal(signal,'signal');
+if(!isStream(stream)){
+thrownewERR_INVALID_ARG_TYPE('stream','stream.Stream',stream);
+}
+constonAbort=()=>{
+stream.destroy(newAbortError());
+};
+if(signal.aborted){
+onAbort();
+}else{
+signal.addEventListener('abort',onAbort);
+eos(stream,()=>signal.removeEventListener('abort',onAbort));
+}
+returnstream;
+};

lib/internal/streams/readable.js

@@ -50,6 +50,7 @@ const{
  getHighWaterMark,
  getDefaultHighWaterMark
 }=require('internal/streams/state');
+
 const{
 ERR_INVALID_ARG_TYPE,
 ERR_STREAM_PUSH_AFTER_EOF,

lib/stream.js

@@ -43,6 +43,7 @@ Stream.Duplex = require('internal/streams/duplex');
 Stream.Transform=require('internal/streams/transform');
 Stream.PassThrough=require('internal/streams/passthrough');
 Stream.pipeline=pipeline;
+Stream.addAbortSignal=require('internal/streams/add-abort-signal');
 Stream.finished=eos;
 
 functionlazyLoadPromises(){

node.gyp

@@ -245,6 +245,7 @@
 'lib/internal/worker/js_transferable.js',
 'lib/internal/watchdog.js',
 'lib/internal/streams/lazy_transform.js',
+'lib/internal/streams/add-abort-signal.js',
 'lib/internal/streams/buffer_list.js',
 'lib/internal/streams/duplexpair.js',
 'lib/internal/streams/from.js',

test/parallel/test-bootstrap-modules.js

@@ -78,6 +78,7 @@ const expectedModules = new Set([
 'NativeModule internal/process/warning',
 'NativeModule internal/querystring',
 'NativeModule internal/source_map/source_map_cache',
+'NativeModule internal/streams/add-abort-signal',
 'NativeModule internal/streams/buffer_list',
 'NativeModule internal/streams/destroy',
 'NativeModule internal/streams/duplex',

test/parallel/test-stream-pipeline.js

@@ -8,7 +8,8 @@ const{
  Transform,
  pipeline,
  PassThrough,
- Duplex
+ Duplex,
+ addAbortSignal,
 }=require('stream');
 constassert=require('assert');
 consthttp=require('http');
@@ -1261,3 +1262,32 @@ const net = require('net');
 ()=>common.mustNotCall(),
 );
 }
+
+
+{
+constac=newAbortController();
+constr=Readable.from(asyncfunction*(){
+for(leti=0;i<10;i++){
+awaitPromise.resolve();
+yieldString(i);
+if(i===5){
+ac.abort();
+}
+}
+}());
+letres='';
+constw=newWritable({
+write(chunk,encoding,callback){
+res+=chunk;
+callback();
+}
+});
+constcb=common.mustCall((err)=>{
+assert.strictEqual(err.name,'AbortError');
+assert.strictEqual(res,'012345');
+assert.strictEqual(w.destroyed,true);
+assert.strictEqual(r.destroyed,true);
+assert.strictEqual(pipelined.destroyed,true);
+});
+constpipelined=addAbortSignal(ac.signal,pipeline([r,w],cb));
+}

test/parallel/test-stream-readable-destroy.js

@@ -1,7 +1,7 @@
 'use strict';
 
 constcommon=require('../common');
-const{ Readable }=require('stream');
+const{ Readable, addAbortSignal}=require('stream');
 constassert=require('assert');
 
 {
@@ -268,3 +268,38 @@ const assert = require('assert');
 }));
 read.resume();
 }
+
+{
+constcontroller=newAbortController();
+constread=addAbortSignal(controller.signal,newReadable({
+read(){
+this.push('asd');
+},
+}));
+
+read.on('error',common.mustCall((e)=>{
+assert.strictEqual(e.name,'AbortError');
+}));
+controller.abort();
+read.on('data',common.mustNotCall());
+}
+
+{
+constcontroller=newAbortController();
+constread=addAbortSignal(controller.signal,newReadable({
+objectMode: true,
+read(){
+returnfalse;
+}
+}));
+read.push('asd');
+
+read.on('error',common.mustCall((e)=>{
+assert.strictEqual(e.name,'AbortError');
+}));
+assert.rejects((async()=>{
+/* eslint-disable-next-line no-unused-vars */
+forawait(constchunkofread){}
+})(),/AbortError/);
+setTimeout(()=>controller.abort(),0);
+}

test/parallel/test-stream-writable-destroy.js

@@ -1,7 +1,7 @@
 'use strict';
 
 constcommon=require('../common');
-const{ Writable }=require('stream');
+const{ Writable, addAbortSignal}=require('stream');
 constassert=require('assert');
 
 {
@@ -417,3 +417,17 @@ const assert = require('assert');
 }));
 write.write('asd');
 }
+
+{
+constac=newAbortController();
+constwrite=addAbortSignal(ac.signal,newWritable({
+write(chunk,enc,cb){cb();}
+}));
+
+write.on('error',common.mustCall((e)=>{
+assert.strictEqual(e.name,'AbortError');
+assert.strictEqual(write.destroyed,true);
+}));
+write.write('asd');
+ac.abort();
+}