stream: support dispose in writable

Author: benjamingr

lib/internal/streams/writable.js

@@ -34,6 +34,7 @@ const {
   ObjectSetPrototypeOf,
   StringPrototypeToLowerCase,
   Symbol,
+  SymbolAsyncDispose,
   SymbolHasInstance,
 } = primordials;
 
@@ -44,6 +45,7 @@ const EE = require('events');
 const Stream = require('internal/streams/legacy').Stream;
 const { Buffer } = require('buffer');
 const destroyImpl = require('internal/streams/destroy');
+const eos = require('internal/streams/end-of-stream');
 
 const {
   addAbortSignal,
@@ -54,16 +56,19 @@ const {
   getDefaultHighWaterMark,
 } = require('internal/streams/state');
 const {
-  ERR_INVALID_ARG_TYPE,
-  ERR_METHOD_NOT_IMPLEMENTED,
-  ERR_MULTIPLE_CALLBACK,
-  ERR_STREAM_CANNOT_PIPE,
-  ERR_STREAM_DESTROYED,
-  ERR_STREAM_ALREADY_FINISHED,
-  ERR_STREAM_NULL_VALUES,
-  ERR_STREAM_WRITE_AFTER_END,
-  ERR_UNKNOWN_ENCODING,
-} = require('internal/errors').codes;
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_METHOD_NOT_IMPLEMENTED,
+    ERR_MULTIPLE_CALLBACK,
+    ERR_STREAM_CANNOT_PIPE,
+    ERR_STREAM_DESTROYED,
+    ERR_STREAM_ALREADY_FINISHED,
+    ERR_STREAM_NULL_VALUES,
+    ERR_STREAM_WRITE_AFTER_END,
+    ERR_UNKNOWN_ENCODING,
+  },
+} = require('internal/errors');
 
 const { errorOrDestroy } = destroyImpl;
 
@@ -477,7 +482,7 @@ function onwrite(stream, er) {
       // rather just increase a counter, to improve performance and avoid
       // memory allocations.
       if (state.afterWriteTickInfo !== null &&
-          state.afterWriteTickInfo.cb === cb) {
+        state.afterWriteTickInfo.cb === cb) {
         state.afterWriteTickInfo.count++;
       } else {
         state.afterWriteTickInfo = { count: 1, cb, stream, state };
@@ -538,9 +543,9 @@ function errorBuffer(state) {
 // If there's something in the buffer waiting, then process it.
 function clearBuffer(stream, state) {
   if (state.corked ||
-      state.bufferProcessing ||
-      state.destroyed ||
-      !state.constructed) {
+    state.bufferProcessing ||
+    state.destroyed ||
+    !state.constructed) {
     return;
   }
 
@@ -661,15 +666,15 @@ Writable.prototype.end = function(chunk, encoding, cb) {
 
 function needFinish(state) {
   return (state.ending &&
-          !state.destroyed &&
-          state.constructed &&
-          state.length === 0 &&
-          !state.errored &&
-          state.buffered.length === 0 &&
-          !state.finished &&
-          !state.writing &&
-          !state.errorEmitted &&
-          !state.closeEmitted);
+    !state.destroyed &&
+    state.constructed &&
+    state.length === 0 &&
+    !state.errored &&
+    state.buffered.length === 0 &&
+    !state.finished &&
+    !state.writing &&
+    !state.errorEmitted &&
+    !state.closeEmitted);
 }
 
 function callFinal(stream, state) {
@@ -934,3 +939,12 @@ Writable.fromWeb = function(writableStream, options) {
 Writable.toWeb = function(streamWritable) {
   return lazyWebStreams().newWritableStreamFromStreamWritable(streamWritable);
 };
+
+Writable.prototype[SymbolAsyncDispose] = function() {
+  let error;
+  if (!this.destroyed) {
+    error = this.readableEnded ? null : new AbortError();
+    this.destroy(error);
+  }
+  return new Promise((resolve, reject) => eos(this, (err) => (err && err !== error ? reject(err) : resolve(null))));
+};

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

@@ -255,3 +255,17 @@ const assert = require('assert');
   duplex.on('close', common.mustCall());
   controller.abort();
 }
+{
+  // Check Symbol.asyncDispose
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {},
+  });
+  let count = 0;
+  duplex.on('error', common.mustCall((e) => {
+    assert.strictEqual(count++, 0); // Ensure not called twice
+    assert.strictEqual(e.name, 'AbortError');
+  }));
+  duplex.on('close', common.mustCall());
+  duplex[Symbol.asyncDispose]().then(common.mustCall());
+}

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

@@ -141,3 +141,14 @@ const assert = require('assert');
 
   transform.destroy();
 }
+
+{
+  const transform = new Transform({
+    transform(chunk, enc, cb) {}
+  });
+  transform.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.name, "AbortError");
+  }));
+  transform.on('clocse', common.mustCall());
+  transform[Symbol.asyncDispose]().then(common.mustCall());
+}

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

@@ -487,3 +487,16 @@ const assert = require('assert');
   }));
   s.destroy(_err);
 }
+
+{
+
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(); }
+  });
+
+  write.on('error', common.mustCall((e) => {
+    assert.strictEqual(e.name, 'AbortError');
+    assert.strictEqual(write.destroyed, true);
+  }));
+  write[Symbol.asyncDispose]().then(common.mustCall());
+}