stream: use internal addAbortListener

atlowChemi · marco-ippolito · commit 8442457117c5 · 2024-05-03T08:29:52.000+02:00
PR-URL: #52081 Refs: #48596 Reviewed-By: Moshe Atlow <moshe@atlow.co.il> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com> Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
diff --git a/lib/internal/streams/add-abort-signal.js b/lib/internal/streams/add-abort-signal.js
@@ -51,7 +51,7 @@ module.exports.addAbortSignalNoValidate = function(signal, stream) {
   if (signal.aborted) {
     onAbort();
   } else {
-    addAbortListener ??= require('events').addAbortListener;
+    addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
     const disposable = addAbortListener(signal, onAbort);
     eos(stream, disposable[SymbolDispose]);
   }
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
@@ -254,7 +254,7 @@ function eos(stream, options, callback) {
     if (options.signal.aborted) {
       process.nextTick(abort);
     } else {
-      addAbortListener ??= require('events').addAbortListener;
+      addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
       const disposable = addAbortListener(options.signal, abort);
       const originalCallback = callback;
       callback = once((...args) => {
@@ -278,7 +278,7 @@ function eosWeb(stream, options, callback) {
     if (options.signal.aborted) {
       process.nextTick(abort);
     } else {
-      addAbortListener ??= require('events').addAbortListener;
+      addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
       const disposable = addAbortListener(options.signal, abort);
       const originalCallback = callback;
       callback = once((...args) => {
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -208,7 +208,7 @@ function pipelineImpl(streams, callback, opts) {
     finishImpl(new AbortError());
   }
 
-  addAbortListener ??= require('events').addAbortListener;
+  addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
   let disposable;
   if (outerSignal) {
     disposable = addAbortListener(outerSignal, abort);
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
@@ -1493,7 +1493,7 @@ function readableStreamPipeTo(
       abortAlgorithm();
       return promise.promise;
     }
-    addAbortListener ??= require('events').addAbortListener;
+    addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
     disposable = addAbortListener(signal, abortAlgorithm);
   }
 
