stream: fix destroy(err, cb) regression

lib/internal/streams/destroy.js

@@ -8,7 +8,10 @@ function destroy(err, cb) {
     this._writableState.destroyed;
 
   if (readableDestroyed || writableDestroyed) {
-    if (err && (!this._writableState || !this._writableState.errorEmitted)) {
+    if (cb) {
+      cb(err);
+    } else if (err &&
+               (!this._writableState || !this._writableState.errorEmitted)) {
       process.nextTick(emitErrorNT, this, err);
     }
     return;

test/parallel/test-net-socket-destroy-send.js

@@ -0,0 +1,22 @@
+'use strict';
+
+const common = require('../common');
+const net = require('net');
+const assert = require('assert');
+
+const server = net.createServer();
+server.listen(0, common.mustCall(function() {
+  const port = server.address().port;
+  const conn = net.createConnection(port);
+
+  conn.on('connect', common.mustCall(function() {
+    conn.destroy();
+    conn.on('error', common.mustCall(function(err) {
+      assert.strictEqual(err.message, 'This socket is closed');
+    }));
+    conn.write(Buffer.from('kaboom'), common.mustCall(function(err) {
+      assert.strictEqual(err.message, 'This socket is closed');
+    }));
+    server.close();
+  }));
+}));

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

@@ -160,3 +160,17 @@ const { inherits } = require('util');
 
   new MyReadable();
 }
+
+{
+  // destroy and destroy callback
+  const read = new Readable({
+    read() {}
+  });
+  read.resume();
+
+  const expected = new Error('kaboom');
+
+  read.destroy(expected, common.mustCall(function(err) {
+    assert.strictEqual(expected, err);
+  }));
+}

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

@@ -170,3 +170,18 @@ const { inherits } = require('util');
 
   new MyWritable();
 }
+
+{
+  // destroy and destroy callback
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(); }
+  });
+
+  write.destroy();
+
+  const expected = new Error('kaboom');
+
+  write.destroy(expected, common.mustCall(function(err) {
+    assert.strictEqual(expected, err);
+  }));
+}