stream: make stream.destroy with callback API public

This makes the `stream.destroy(err, callback)` API public. Additionally it makes some changes for easier use: - The callback is always invoked with the same behavior as eos. - The error is assumed to be handled and uncaughException is supressed. - The callback timing is the same regardless whether destroy has already been called or not. - The callback is always invoked asynchronously. - The callback used be invoked before emitting 'error' and/or 'close'.
nodejs · ronag · Feb 29, 2020 · Feb 29, 2020 · Feb 29, 2020 · Feb 29, 2020
commit 4d51ce200dc852149646705790e61d9d9e177711
diff --git a/doc/api/stream.md b/doc/api/stream.md
@@ -369,7 +369,7 @@ to be processed. However, use of `writable.cork()` without implementing
 
 See also: [`writable.uncork()`][], [`writable._writev()`][stream-_writev].
 
-##### `writable.destroy([error])`
+##### `writable.destroy([error, callback])`
 <!-- YAML
 added: v8.0.0
 -->
@@ -389,6 +389,11 @@ the `'drain'` event before destroying the stream.
 Once `destroy()` has been called any further calls will be a noop and no
 further errors except from `_destroy` may be emitted as `'error'`.
 
+If passed `callback`; it will be invoked once the stream destrution has
+completed. If an error has occured it will be passed as the first argument to
+the callback and no `uncaughtException` error will occur even if no `'error'`
+listener has been registered on the stream.
+
 Implementors should not override this method,
 but instead implement [`writable._destroy()`][writable-_destroy].
 

diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
@@ -1,18 +1,24 @@
 'use strict';
 
-// Undocumented cb() API, needed for core, not for public API.
-// The cb() will be invoked synchronously if _destroy is synchronous.
-// If cb is passed no 'error' event will be emitted.
+let eos;
+
 function destroy(err, cb) {
   const r = this._readableState;
   const w = this._writableState;
 
-  if ((w && w.destroyed) || (r && r.destroyed)) {
-    if (typeof cb === 'function') {
-      // TODO(ronag): Invoke with `'close'`/`'error'`.
-      cb();
-    }
+  if (typeof err === 'function') {
+    cb = err;
+    err = null;
+  }
 
+  if (typeof cb === 'function') {
+    if (!eos) eos = require('internal/streams/end-of-stream');
+    eos(this, (err) => {
+      cb(err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE' ? err : undefined);
+    });
+  }
+
+  if ((w && w.destroyed) || (r && r.destroyed)) {
     return this;
   }
 
@@ -52,10 +58,6 @@ function destroy(err, cb) {
       r.closed = true;
     }
 
-    if (typeof cb === 'function') {
-      cb(err);
-    }
-
     if (err) {
       process.nextTick(emitErrorCloseNT, this, err);
     } else {

diff --git a/test/parallel/test-stream-writable-destroy.js b/test/parallel/test-stream-writable-destroy.js
@@ -248,9 +248,33 @@ const assert = require('assert');
 
   const expected = new Error('kaboom');
 
+  let ticked = false;
+  write.destroy(expected, common.mustCall((err) => {
+    assert.strictEqual(err, undefined);
+    assert.strictEqual(ticked, true);
+    let ticked2 = false;
+    write.destroy(expected, common.mustCall((err) => {
+      assert.strictEqual(err, undefined);
+      assert.strictEqual(ticked2, true);
+    }));
+    ticked2 = true;
+  }));
+  ticked = true;
+
+  // Destroy already destroyed.
+
+  ticked = false;
   write.destroy(expected, common.mustCall((err) => {
     assert.strictEqual(err, undefined);
+    assert.strictEqual(ticked, true);
+    let ticked2 = false;
+    write.destroy(expected, common.mustCall((err) => {
+      assert.strictEqual(err, undefined);
+      assert.strictEqual(ticked2, true);
+    }));
+    ticked2 = true;
   }));
+  ticked = true;
 }
 
 {