stream: add auto-destroy mode

PR-URL: #22795
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Jeremiah Senkpiel <[email protected]>

Author: mafintosh

doc/api/stream.md

@@ -1503,6 +1503,11 @@ changes:
     pr-url: https://github.com/nodejs/node/pull/18438
     description: >
       Add `emitClose` option to specify if `'close'` is emitted on destroy
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/22795
+    description: >
+      Add `autoDestroy` option to automatically `destroy()` the stream
+      when it emits `'finish'` or errors
 -->
 
 * `options` {Object}
@@ -1531,6 +1536,8 @@ changes:
     [`stream._destroy()`][writable-_destroy] method.
   * `final` {Function} Implementation for the
     [`stream._final()`][stream-_final] method.
+  * `autoDestroy` {boolean} Whether this stream should automatically call
+    `.destroy()` on itself after ending. **Default:** `false`.
 
 <!-- eslint-disable no-useless-constructor -->
 ```js
@@ -1762,6 +1769,14 @@ Custom `Readable` streams *must* call the `new stream.Readable([options])`
 constructor and implement the `readable._read()` method.
 
 #### new stream.Readable([options])
+<!-- YAML
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/22795
+    description: >
+      Add `autoDestroy` option to automatically `destroy()` the stream
+      when it emits `'end'` or errors
+-->
 
 * `options` {Object}
   * `highWaterMark` {number} The maximum [number of bytes][hwm-gotcha] to store
@@ -1776,6 +1791,8 @@ constructor and implement the `readable._read()` method.
     method.
   * `destroy` {Function} Implementation for the
     [`stream._destroy()`][readable-_destroy] method.
+  * `autoDestroy` {boolean} Whether this stream should automatically call
+    `.destroy()` on itself after ending. **Default:** `false`.
 
 <!-- eslint-disable no-useless-constructor -->
 ```js

lib/_stream_readable.js

@@ -46,6 +46,7 @@ let createReadableStreamAsyncIterator;
 
 util.inherits(Readable, Stream);
 
+const { errorOrDestroy } = destroyImpl;
 const kProxyEvents = ['error', 'close', 'destroy', 'pause', 'resume'];
 
 function prependListener(emitter, event, fn) {
@@ -118,6 +119,9 @@ function ReadableState(options, stream, isDuplex) {
   // Should close be emitted on destroy. Defaults to true.
   this.emitClose = options.emitClose !== false;
 
+  // Should .destroy() be called after 'end' (and potentially 'finish')
+  this.autoDestroy = !!options.autoDestroy;
+
   // has it been destroyed
   this.destroyed = false;
 
@@ -236,7 +240,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
     if (!skipChunkCheck)
       er = chunkInvalid(state, chunk);
     if (er) {
-      stream.emit('error', er);
+      errorOrDestroy(stream, er);
     } else if (state.objectMode || chunk && chunk.length > 0) {
       if (typeof chunk !== 'string' &&
           !state.objectMode &&
@@ -246,11 +250,11 @@ function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
 
       if (addToFront) {
         if (state.endEmitted)
-          stream.emit('error', new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());
+          errorOrDestroy(stream, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());
         else
           addChunk(stream, state, chunk, true);
       } else if (state.ended) {
-        stream.emit('error', new ERR_STREAM_PUSH_AFTER_EOF());
+        errorOrDestroy(stream, new ERR_STREAM_PUSH_AFTER_EOF());
       } else if (state.destroyed) {
         return false;
       } else {
@@ -604,7 +608,7 @@ function maybeReadMore_(stream, state) {
 // for virtual (non-string, non-buffer) streams, "length" is somewhat
 // arbitrary, and perhaps not very meaningful.
 Readable.prototype._read = function(n) {
-  this.emit('error', new ERR_METHOD_NOT_IMPLEMENTED('_read()'));
+  errorOrDestroy(this, new ERR_METHOD_NOT_IMPLEMENTED('_read()'));
 };
 
 Readable.prototype.pipe = function(dest, pipeOpts) {
@@ -710,7 +714,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     unpipe();
     dest.removeListener('error', onerror);
     if (EE.listenerCount(dest, 'error') === 0)
-      dest.emit('error', er);
+      errorOrDestroy(dest, er);
   }
 
   // Make sure our error handler is attached before userland ones.
@@ -1123,5 +1127,14 @@ function endReadableNT(state, stream) {
     state.endEmitted = true;
     stream.readable = false;
     stream.emit('end');
+
+    if (state.autoDestroy) {
+      // In case of duplex streams we need a way to detect
+      // if the writable side is ready for autoDestroy as well
+      const wState = stream._writableState;
+      if (!wState || (wState.autoDestroy && wState.finished)) {
+        stream.destroy();
+      }
+    }
   }
 }

lib/_stream_writable.js

@@ -45,6 +45,8 @@ const {
   ERR_UNKNOWN_ENCODING
 } = require('internal/errors').codes;
 
+const { errorOrDestroy } = destroyImpl;
+
 util.inherits(Writable, Stream);
 
 function nop() {}
@@ -147,6 +149,9 @@ function WritableState(options, stream, isDuplex) {
   // Should close be emitted on destroy. Defaults to true.
   this.emitClose = options.emitClose !== false;
 
+  // Should .destroy() be called after 'finish' (and potentially 'end')
+  this.autoDestroy = !!options.autoDestroy;
+
   // count buffered requests
   this.bufferedRequestCount = 0;
 
@@ -235,14 +240,14 @@ function Writable(options) {
 
 // Otherwise people can pipe Writable streams, which is just wrong.
 Writable.prototype.pipe = function() {
-  this.emit('error', new ERR_STREAM_CANNOT_PIPE());
+  errorOrDestroy(this, new ERR_STREAM_CANNOT_PIPE());
 };
 
 
 function writeAfterEnd(stream, cb) {
   var er = new ERR_STREAM_WRITE_AFTER_END();
   // TODO: defer error events consistently everywhere, not just the cb
-  stream.emit('error', er);
+  errorOrDestroy(stream, er);
   process.nextTick(cb, er);
 }
 
@@ -258,7 +263,7 @@ function validChunk(stream, state, chunk, cb) {
     er = new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer'], chunk);
   }
   if (er) {
-    stream.emit('error', er);
+    errorOrDestroy(stream, er);
     process.nextTick(cb, er);
     return false;
   }
@@ -422,13 +427,13 @@ function onwriteError(stream, state, sync, er, cb) {
     // after error
     process.nextTick(finishMaybe, stream, state);
     stream._writableState.errorEmitted = true;
-    stream.emit('error', er);
+    errorOrDestroy(stream, er);
   } else {
     // the caller expect this to happen before if
     // it is async
     cb(er);
     stream._writableState.errorEmitted = true;
-    stream.emit('error', er);
+    errorOrDestroy(stream, er);
     // this can emit finish, but finish must
     // always follow error
     finishMaybe(stream, state);
@@ -612,7 +617,7 @@ function callFinal(stream, state) {
   stream._final((err) => {
     state.pendingcb--;
     if (err) {
-      stream.emit('error', err);
+      errorOrDestroy(stream, err);
     }
     state.prefinished = true;
     stream.emit('prefinish');
@@ -639,6 +644,15 @@ function finishMaybe(stream, state) {
     if (state.pendingcb === 0) {
       state.finished = true;
       stream.emit('finish');
+
+      if (state.autoDestroy) {
+        // In case of duplex streams we need a way to detect
+        // if the readable side is ready for autoDestroy as well
+        const rState = stream._readableState;
+        if (!rState || (rState.autoDestroy && rState.endEmitted)) {
+          stream.destroy();
+        }
+      }
     }
   }
   return need;

lib/internal/streams/destroy.js

@@ -82,7 +82,25 @@ function emitErrorNT(self, err) {
   self.emit('error', err);
 }
 
+function errorOrDestroy(stream, err) {
+  // We have tests that rely on errors being emitted
+  // in the same tick, so changing this is semver major.
+  // For now when you opt-in to autoDestroy we allow
+  // the error to be emitted nextTick. In a future
+  // semver major update we should change the default to this.
+
+  const rState = stream._readableState;
+  const wState = stream._writableState;
+
+  if ((rState && rState.autoDestroy) || (wState && wState.autoDestroy))
+    stream.destroy(err);
+  else
+    stream.emit('error', err);
+}
+
+
 module.exports = {
   destroy,
-  undestroy
+  undestroy,
+  errorOrDestroy
 };

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

@@ -0,0 +1,84 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+const assert = require('assert');
+
+{
+  const r = new stream.Readable({
+    autoDestroy: true,
+    read() {
+      this.push('hello');
+      this.push('world');
+      this.push(null);
+    },
+    destroy: common.mustCall((err, cb) => cb())
+  });
+
+  let ended = false;
+
+  r.resume();
+
+  r.on('end', common.mustCall(() => {
+    ended = true;
+  }));
+
+  r.on('close', common.mustCall(() => {
+    assert(ended);
+  }));
+}
+
+{
+  const w = new stream.Writable({
+    autoDestroy: true,
+    write(data, enc, cb) {
+      cb(null);
+    },
+    destroy: common.mustCall((err, cb) => cb())
+  });
+
+  let finished = false;
+
+  w.write('hello');
+  w.write('world');
+  w.end();
+
+  w.on('finish', common.mustCall(() => {
+    finished = true;
+  }));
+
+  w.on('close', common.mustCall(() => {
+    assert(finished);
+  }));
+}
+
+{
+  const t = new stream.Transform({
+    autoDestroy: true,
+    transform(data, enc, cb) {
+      cb(null, data);
+    },
+    destroy: common.mustCall((err, cb) => cb())
+  });
+
+  let ended = false;
+  let finished = false;
+
+  t.write('hello');
+  t.write('world');
+  t.end();
+
+  t.resume();
+
+  t.on('end', common.mustCall(() => {
+    ended = true;
+  }));
+
+  t.on('finish', common.mustCall(() => {
+    finished = true;
+  }));
+
+  t.on('close', common.mustCall(() => {
+    assert(ended);
+    assert(finished);
+  }));
+}