stream: add destroy and _destroy methods.

Adds destroy() and _destroy() methods to Readable, Writable, Duplex
and Transform. It also standardizes the behavior and the implementation
of destroy(), which has been inconsistent in userland and core.
This PR also updates all the subsystems of core to use the new
destroy().

PR-URL: https://github.com/nodejs/node/pull/12925
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Calvin Metcalf <calvin.metcalf@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>

1 files changed, 194 insertions, 0 deletions

diff --git a/test/parallel/test-stream-duplex-destroy.js b/test/parallel/test-stream-duplex-destroy.js
new file mode 100644
index 0000000000..00e334d64b
--- /dev/null
+++ b/test/parallel/test-stream-duplex-destroy.js
@@ -0,0 +1,194 @@
+'use strict';
+
+const common = require('../common');
+const { Duplex } = require('stream');
+const assert = require('assert');
+const { inherits } = require('util');
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {}
+  });
+
+  duplex.resume();
+
+  duplex.on('end', common.mustCall());
+  duplex.on('finish', common.mustCall());
+
+  duplex.destroy();
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {}
+  });
+  duplex.resume();
+
+  const expected = new Error('kaboom');
+
+  duplex.on('end', common.mustCall());
+  duplex.on('finish', common.mustCall());
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err, expected);
+  }));
+
+  duplex.destroy(expected);
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {}
+  });
+
+  duplex._destroy = common.mustCall(function(err, cb) {
+    assert.strictEqual(err, expected);
+    cb(err);
+  });
+
+  const expected = new Error('kaboom');
+
+  duplex.on('finish', common.mustNotCall('no finish event'));
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err, expected);
+  }));
+
+  duplex.destroy(expected);
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const expected = new Error('kaboom');
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {},
+    destroy: common.mustCall(function(err, cb) {
+      assert.strictEqual(err, expected);
+      cb();
+    })
+  });
+  duplex.resume();
+
+  duplex.on('end', common.mustNotCall('no end event'));
+  duplex.on('finish', common.mustNotCall('no finish event'));
+
+  // error is swallowed by the custom _destroy
+  duplex.on('error', common.mustNotCall('no error event'));
+
+  duplex.destroy(expected);
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {}
+  });
+
+  duplex._destroy = common.mustCall(function(err, cb) {
+    assert.strictEqual(err, null);
+    cb();
+  });
+
+  duplex.destroy();
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {}
+  });
+  duplex.resume();
+
+  duplex._destroy = common.mustCall(function(err, cb) {
+    assert.strictEqual(err, null);
+    process.nextTick(() => {
+      this.push(null);
+      this.end();
+      cb();
+    });
+  });
+
+  const fail = common.mustNotCall('no finish or end event');
+
+  duplex.on('finish', fail);
+  duplex.on('end', fail);
+
+  duplex.destroy();
+
+  duplex.removeListener('end', fail);
+  duplex.removeListener('finish', fail);
+  duplex.on('end', common.mustCall());
+  duplex.on('finish', common.mustCall());
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {}
+  });
+
+  const expected = new Error('kaboom');
+
+  duplex._destroy = common.mustCall(function(err, cb) {
+    assert.strictEqual(err, null);
+    cb(expected);
+  });
+
+  duplex.on('finish', common.mustNotCall('no finish event'));
+  duplex.on('end', common.mustNotCall('no end event'));
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err, expected);
+  }));
+
+  duplex.destroy();
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {},
+    allowHalfOpen: true
+  });
+  duplex.resume();
+
+  duplex.on('finish', common.mustCall());
+  duplex.on('end', common.mustCall());
+
+  duplex.destroy();
+  assert.strictEqual(duplex.destroyed, true);
+}
+
+{
+  const duplex = new Duplex({
+    write(chunk, enc, cb) { cb(); },
+    read() {},
+  });
+
+  duplex.destroyed = true;
+  assert.strictEqual(duplex.destroyed, true);
+
+  // the internal destroy() mechanism should not be triggered
+  duplex.on('finish', common.mustNotCall());
+  duplex.on('end', common.mustNotCall());
+  duplex.destroy();
+}
+
+{
+  function MyDuplex() {
+    assert.strictEqual(this.destroyed, false);
+    this.destroyed = false;
+    Duplex.call(this);
+  }
+
+  inherits(MyDuplex, Duplex);
+
+  new MyDuplex();
+}