stream: make finished call the callback if the stream is closed

Make stream.finished callback invoked if stream is already
closed/destroyed.

PR-URL: https://github.com/nodejs/node/pull/28748
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

2 files changed, 243 insertions, 3 deletions

diff --git a/test/parallel/test-http-client-finished.js b/test/parallel/test-http-client-finished.js
index 2d7e5b95b3..337f7b596d 100644
--- a/test/parallel/test-http-client-finished.js
+++ b/test/parallel/test-http-client-finished.js
@@ -25,3 +25,109 @@ const { finished } = require('stream');
     .end();
   }));
 }
+
+{
+  // Test abort before finished.
+
+  const server = http.createServer(function(req, res) {
+  });
+
+  server.listen(0, common.mustCall(function() {
+    const req = http.request({
+      port: this.address().port
+    }, common.mustNotCall());
+    req.abort();
+    finished(req, common.mustCall(() => {
+      server.close();
+    }));
+  }));
+}
+
+{
+  // Test abort after request.
+
+  const server = http.createServer(function(req, res) {
+  });
+
+  server.listen(0, common.mustCall(function() {
+    const req = http.request({
+      port: this.address().port
+    }).end();
+    finished(req, (err) => {
+      common.expectsError({
+        type: Error,
+        code: 'ERR_STREAM_PREMATURE_CLOSE'
+      })(err);
+      finished(req, common.mustCall(() => {
+        server.close();
+      }));
+    });
+    req.abort();
+  }));
+}
+
+{
+  // Test abort before end.
+
+  const server = http.createServer(function(req, res) {
+    res.write('test');
+  });
+
+  server.listen(0, common.mustCall(function() {
+    const req = http.request({
+      port: this.address().port
+    }).on('response', common.mustCall((res) => {
+      req.abort();
+      finished(res, common.mustCall(() => {
+        finished(res, common.mustCall(() => {
+          server.close();
+        }));
+      }));
+    })).end();
+  }));
+}
+
+{
+  // Test destroy before end.
+
+  const server = http.createServer(function(req, res) {
+    res.write('test');
+  });
+
+  server.listen(0, common.mustCall(function() {
+    http.request({
+      port: this.address().port
+    }).on('response', common.mustCall((res) => {
+      // TODO(ronag): Bug? Won't emit 'close' unless read.
+      res.on('data', () => {});
+      res.destroy();
+      finished(res, common.mustCall(() => {
+        finished(res, common.mustCall(() => {
+          server.close();
+        }));
+      }));
+    })).end();
+  }));
+}
+
+{
+  // Test finish after end.
+
+  const server = http.createServer(function(req, res) {
+    res.end('asd');
+  });
+
+  server.listen(0, common.mustCall(function() {
+    http.request({
+      port: this.address().port
+    }).on('response', common.mustCall((res) => {
+      // TODO(ronag): Bug? Won't emit 'close' unless read.
+      res.on('data', () => {});
+      finished(res, common.mustCall(() => {
+        finished(res, common.mustCall(() => {
+          server.close();
+        }));
+      }));
+    })).end();
+  }));
+}
diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js
index d6361ea303..c5e792b45e 100644
--- a/test/parallel/test-stream-finished.js
+++ b/test/parallel/test-stream-finished.js
@@ -101,11 +101,25 @@ const { promisify } = require('util');
   const rs = new Readable();
 
   finished(rs, common.mustCall((err) => {
+    assert(err, 'premature close error');
+  }));
+
+  rs.push(null);
+  rs.emit('close');
+  rs.resume();
+}
+
+{
+  const rs = new Readable();
+
+  finished(rs, common.mustCall((err) => {
     assert(!err, 'no error');
   }));
 
   rs.push(null);
-  rs.emit('close'); // Should not trigger an error
+  rs.on('end', common.mustCall(() => {
+    rs.emit('close'); // Should not trigger an error
+  }));
   rs.resume();
 }
 
@@ -155,8 +169,9 @@ const { promisify } = require('util');
   rs.resume();
 }
 
-// Test that calling returned function removes listeners
 {
+  // Nothing happens if disposed.
+
   const ws = new Writable({
     write(data, env, cb) {
       cb();
@@ -168,6 +183,8 @@ const { promisify } = require('util');
 }
 
 {
+  // Nothing happens if disposed.
+
   const rs = new Readable();
   const removeListeners = finished(rs, common.mustNotCall());
   removeListeners();
@@ -178,9 +195,126 @@ const { promisify } = require('util');
 }
 
 {
+  // Completed if readable-like is ended before.
+
   const streamLike = new EE();
   streamLike.readableEnded = true;
   streamLike.readable = true;
-  finished(streamLike, common.mustCall);
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if readable-like is never ended.
+
+  const streamLike = new EE();
+  streamLike.readableEnded = false;
+  streamLike.readable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.emit('close');
+}
+
+{
+  // Completed if writable-like is destroyed before.
+
+  const streamLike = new EE();
+  streamLike.destroyed = true;
+  streamLike.writable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if readable-like is aborted before.
+
+  const streamLike = new EE();
+  streamLike.destroyed = true;
+  streamLike.readable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if writable-like is aborted before.
+
+  const streamLike = new EE();
+  streamLike.aborted = true;
+  streamLike.writable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if readable-like is aborted before.
+
+  const streamLike = new EE();
+  streamLike.aborted = true;
+  streamLike.readable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if streamlike is finished before.
+
+  const streamLike = new EE();
+  streamLike.writableFinished = true;
+  streamLike.writable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Premature close if stream is not finished.
+
+  const streamLike = new EE();
+  streamLike.writableFinished = false;
+  streamLike.writable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.emit('close');
+}
+
+{
+  // Premature close if stream never emitted 'finish'
+  // even if writableFinished says something else.
+
+  const streamLike = new EE();
+  streamLike.writable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.writableFinished = true;
+  streamLike.emit('close');
+}
+
+
+{
+  // Premature close if stream never emitted 'end'
+  // even if readableEnded says something else.
+
+  const streamLike = new EE();
+  streamLike.readable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.readableEnded = true;
   streamLike.emit('close');
 }
+
+{
+  // Completes if already finished.
+
+  const w = new Writable();
+  finished(w, common.mustCall(() => {
+    finished(w, common.mustCall());
+  }));
+  w.destroy();
+}
+
+{
+  // Completes if already ended.
+
+  const r = new Readable();
+  finished(r, common.mustCall(() => {
+    finished(r, common.mustCall());
+  }));
+  r.destroy();
+}