stream: support abort signal

PR-URL: https://github.com/nodejs/node/pull/36061
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>

4 files changed, 47 insertions, 8 deletions

diff --git a/lib/_http_client.js b/lib/_http_client.js
index 75d36f19e6..6fb5dd65cb 100644
--- a/lib/_http_client.js
+++ b/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,7 +61,6 @@ const {
 } = codes;
 const {
   validateInteger,
-  validateAbortSignal,
 } = require('internal/validators');
 const { getTimerDuration } = require('internal/timers');
 const {
@@ -69,6 +68,8 @@ const {
   DTRACE_HTTP_CLIENT_RESPONSE
 } = require('internal/dtrace');
 
+const { addAbortSignal } = require('stream');
+
 const INVALID_PATH_REGEX = /[^\u0021-\u00ff]/;
 const kError = Symbol('kError');
 
@@ -174,12 +175,7 @@ function ClientRequest(input, options, cb) {
 
   const signal = options.signal;
   if (signal) {
-    validateAbortSignal(signal, 'options.signal');
-    const listener = (e) => this.destroy(new AbortError());
-    signal.addEventListener('abort', listener);
-    this.once('close', () => {
-      signal.removeEventListener('abort', listener);
-    });
+    addAbortSignal(signal, this);
   }
   let method = options.method;
   const methodIsString = (typeof method === 'string');
diff --git a/lib/internal/streams/add-abort-signal.js b/lib/internal/streams/add-abort-signal.js
new file mode 100644
index 0000000000..27fefe96d2
--- /dev/null
+++ b/lib/internal/streams/add-abort-signal.js
@@ -0,0 +1,41 @@
+'use strict';
+
+const {
+  AbortError,
+  codes,
+} = require('internal/errors');
+
+const eos = 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
+const validateAbortSignal = (signal, name) => {
+  if (signal !== undefined &&
+      (signal === null ||
+       typeof signal !== 'object' ||
+       !('aborted' in signal))) {
+    throw new ERR_INVALID_ARG_TYPE(name, 'AbortSignal', signal);
+  }
+};
+
+function isStream(obj) {
+  return !!(obj && typeof obj.pipe === 'function');
+}
+
+module.exports = function addAbortSignal(signal, stream) {
+  validateAbortSignal(signal, 'signal');
+  if (!isStream(stream)) {
+    throw new ERR_INVALID_ARG_TYPE('stream', 'stream.Stream', stream);
+  }
+  const onAbort = () => {
+    stream.destroy(new AbortError());
+  };
+  if (signal.aborted) {
+    onAbort();
+  } else {
+    signal.addEventListener('abort', onAbort);
+    eos(stream, () => signal.removeEventListener('abort', onAbort));
+  }
+  return stream;
+};
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 93153908fe..105caa1e15 100644
--- a/lib/internal/streams/readable.js
+++ b/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,
diff --git a/lib/stream.js b/lib/stream.js
index 11f5ca997f..fb6cf416cd 100644
--- a/lib/stream.js
+++ b/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;
 
 function lazyLoadPromises() {