stream: don't emit 'end' after 'error'

Refs: nodejs#6083

lib/_stream_readable.js

@@ -1214,7 +1214,7 @@ function endReadableNT(state, stream) {
   debug('endReadableNT', state.endEmitted, state.length);
 
   // Check that we didn't get one last unshift.
-  if (!state.endEmitted && state.length === 0) {
+  if (!state.errorEmitted && !state.endEmitted && state.length === 0) {
     state.endEmitted = true;
     stream.readable = false;
     stream.emit('end');

lib/internal/streams/destroy.js

@@ -16,6 +16,8 @@ function destroy(err, cb) {
     }
   }
 
+  // TODO(ronag): readable & writable = false?
+
   if ((w && w.destroyed) || (r && r.destroyed)) {
     if (cb) {
       cb(err);

test/parallel/test-stream-readable-error-end.js

@@ -0,0 +1,26 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+
+{
+  // Ensure 'end' is not emitted after 'error'.
+  // This test is slightly more complicated than
+  // needed in order to better illustrate the invariant.
+
+  const r = new Readable({ read() {} });
+
+  let errorEmitted = false;
+
+  r.on('data', common.mustCall());
+  r.on('end', () => {
+    assert.strictEqual(!errorEmitted);
+  });
+  r.on('error', () => {
+    errorEmitted = true;
+  });
+  r.push('asd');
+  r.push(null);
+  r.destroy(new Error('kaboom'));
+}