stream: finished callback for closed streams

Previously finished(stream, cb) would not invoke the callback
for streams that have already finished, ended or errored
before being passed to finished(stream, cb).

PR-URL: https://github.com/nodejs/node/pull/31509
Refs: https://github.com/nodejs/node/pull/31508
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

1 files changed, 117 insertions, 0 deletions

diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js
index f6515a01b8..e866ba3d74 100644
--- a/test/parallel/test-stream-finished.js
+++ b/test/parallel/test-stream-finished.js
@@ -215,3 +215,120 @@ const { promisify } = require('util');
   w.end('asd');
   w.destroy();
 }
+
+function testClosed(factory) {
+  {
+    // If already destroyed but finished is cancelled in same tick
+    // don't invoke the callback,
+
+    const s = factory();
+    s.destroy();
+    const dispose = finished(s, common.mustNotCall());
+    dispose();
+  }
+
+  {
+    // If already destroyed invoked callback.
+
+    const s = factory();
+    s.destroy();
+    finished(s, common.mustCall());
+  }
+
+  {
+    // Don't invoke until destroy has completed.
+
+    let destroyed = false;
+    const s = factory({
+      destroy(err, cb) {
+        setImmediate(() => {
+          destroyed = true;
+          cb();
+        });
+      }
+    });
+    s.destroy();
+    finished(s, common.mustCall(() => {
+      assert.strictEqual(destroyed, true);
+    }));
+  }
+
+  {
+    // Invoke callback even if close is inhibited.
+
+    const s = factory({
+      emitClose: false,
+      destroy(err, cb) {
+        cb();
+        finished(s, common.mustCall());
+      }
+    });
+    s.destroy();
+  }
+
+  {
+    // Invoke with deep async.
+
+    const s = factory({
+      destroy(err, cb) {
+        setImmediate(() => {
+          cb();
+          setImmediate(() => {
+            finished(s, common.mustCall());
+          });
+        });
+      }
+    });
+    s.destroy();
+  }
+}
+
+testClosed((opts) => new Readable({ ...opts }));
+testClosed((opts) => new Writable({ write() {}, ...opts }));
+
+{
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb();
+    },
+    autoDestroy: false
+  });
+  w.end('asd');
+  process.nextTick(() => {
+    finished(w, common.mustCall());
+  });
+}
+
+{
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb(new Error());
+    },
+    autoDestroy: false
+  });
+  w.write('asd');
+  w.on('error', common.mustCall(() => {
+    finished(w, common.mustCall());
+  }));
+}
+
+
+{
+  const r = new Readable({
+    autoDestroy: false
+  });
+  r.push(null);
+  r.resume();
+  r.on('end', common.mustCall(() => {
+    finished(r, common.mustCall());
+  }));
+}
+
+{
+  const rs = fs.createReadStream(__filename, { autoClose: false });
+  rs.resume();
+  rs.on('close', common.mustNotCall());
+  rs.on('end', common.mustCall(() => {
+    finished(rs, common.mustCall());
+  }));
+}