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, 83 insertions, 3 deletions

diff --git a/test/parallel/test-bootstrap-modules.js b/test/parallel/test-bootstrap-modules.js
index a6f2c15c19..0887b8a483 100644
--- a/test/parallel/test-bootstrap-modules.js
+++ b/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',
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 23887122fc..5b12531f92 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/test/parallel/test-stream-pipeline.js
@@ -8,7 +8,8 @@ const {
   Transform,
   pipeline,
   PassThrough,
-  Duplex
+  Duplex,
+  addAbortSignal,
 } = require('stream');
 const assert = require('assert');
 const http = require('http');
@@ -1261,3 +1262,32 @@ const net = require('net');
     () => common.mustNotCall(),
   );
 }
+
+
+{
+  const ac = new AbortController();
+  const r = Readable.from(async function* () {
+    for (let i = 0; i < 10; i++) {
+      await Promise.resolve();
+      yield String(i);
+      if (i === 5) {
+        ac.abort();
+      }
+    }
+  }());
+  let res = '';
+  const w = new Writable({
+    write(chunk, encoding, callback) {
+      res += chunk;
+      callback();
+    }
+  });
+  const cb = 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);
+  });
+  const pipelined = addAbortSignal(ac.signal, pipeline([r, w], cb));
+}
diff --git a/test/parallel/test-stream-readable-destroy.js b/test/parallel/test-stream-readable-destroy.js
index 8ab78ec8cc..1e765f3697 100644
--- a/test/parallel/test-stream-readable-destroy.js
+++ b/test/parallel/test-stream-readable-destroy.js
@@ -1,7 +1,7 @@
 'use strict';
 
 const common = require('../common');
-const { Readable } = require('stream');
+const { Readable, addAbortSignal } = require('stream');
 const assert = require('assert');
 
 {
@@ -268,3 +268,38 @@ const assert = require('assert');
   }));
   read.resume();
 }
+
+{
+  const controller = new AbortController();
+  const read = addAbortSignal(controller.signal, new Readable({
+    read() {
+      this.push('asd');
+    },
+  }));
+
+  read.on('error', common.mustCall((e) => {
+    assert.strictEqual(e.name, 'AbortError');
+  }));
+  controller.abort();
+  read.on('data', common.mustNotCall());
+}
+
+{
+  const controller = new AbortController();
+  const read = addAbortSignal(controller.signal, new Readable({
+    objectMode: true,
+    read() {
+      return false;
+    }
+  }));
+  read.push('asd');
+
+  read.on('error', common.mustCall((e) => {
+    assert.strictEqual(e.name, 'AbortError');
+  }));
+  assert.rejects((async () => {
+    /* eslint-disable-next-line no-unused-vars */
+    for await (const chunk of read) {}
+  })(), /AbortError/);
+  setTimeout(() => controller.abort(), 0);
+}
diff --git a/test/parallel/test-stream-writable-destroy.js b/test/parallel/test-stream-writable-destroy.js
index dca01724f7..0fcaf89289 100644
--- a/test/parallel/test-stream-writable-destroy.js
+++ b/test/parallel/test-stream-writable-destroy.js
@@ -1,7 +1,7 @@
 'use strict';
 
 const common = require('../common');
-const { Writable } = require('stream');
+const { Writable, addAbortSignal } = require('stream');
 const assert = require('assert');
 
 {
@@ -417,3 +417,17 @@ const assert = require('assert');
   }));
   write.write('asd');
 }
+
+{
+  const ac = new AbortController();
+  const write = addAbortSignal(ac.signal, new Writable({
+    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();
+}