zlib: align with streams

- Ensure automatic destruction only happens after both
'end' and 'finish' has been emitted through autoDestroy.
- Ensure close() callback is always invoked.
- Ensure 'error' is only emitted once.

PR-URL: #32220
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

lib/zlib.js

@@ -55,13 +55,15 @@ const {
 } = require('internal/util/types');
 const binding = internalBinding('zlib');
 const assert = require('internal/assert');
+const finished = require('internal/streams/end-of-stream');
 const {
   Buffer,
   kMaxLength
 } = require('buffer');
 const { owner_symbol } = require('internal/async_hooks').symbols;
 
 const kFlushFlag = Symbol('kFlushFlag');
+const kError = Symbol('kError');
 
 const constants = internalBinding('constants').zlib;
 const {
@@ -173,14 +175,13 @@ function zlibOnError(message, errno, code) {
   const self = this[owner_symbol];
   // There is no way to cleanly recover.
   // Continuing only obscures problems.
-  _close(self);
-  self._hadError = true;
 
   // eslint-disable-next-line no-restricted-syntax
   const error = new Error(message);
   error.errno = errno;
   error.code = code;
-  self.emit('error', error);
+  self.destroy(error);
+  self[kError] = error;
 }
 
 // 1. Returns false for undefined and NaN
@@ -260,8 +261,8 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) {
     }
   }
 
-  Transform.call(this, { autoDestroy: false, ...opts });
-  this._hadError = false;
+  Transform.call(this, { autoDestroy: true, ...opts });
+  this[kError] = null;
   this.bytesWritten = 0;
   this._handle = handle;
   handle[owner_symbol] = this;
@@ -274,7 +275,6 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) {
   this._defaultFlushFlag = flush;
   this._finishFlushFlag = finishFlush;
   this._defaultFullFlushFlag = fullFlush;
-  this.once('end', this.close);
   this._info = opts && opts.info;
 }
 ObjectSetPrototypeOf(ZlibBase.prototype, Transform.prototype);
@@ -368,7 +368,7 @@ ZlibBase.prototype.flush = function(kind, callback) {
 };
 
 ZlibBase.prototype.close = function(callback) {
-  _close(this, callback);
+  if (callback) finished(this, callback);
   this.destroy();
 };
 
@@ -432,6 +432,8 @@ function processChunkSync(self, chunk, flushFlag) {
                      availOutBefore); // out_len
     if (error)
       throw error;
+    else if (self[kError])
+      throw self[kError];
 
     availOutAfter = state[0];
     availInAfter = state[1];
@@ -512,7 +514,7 @@ function processCallback() {
   const self = this[owner_symbol];
   const state = self._writeState;
 
-  if (self._hadError || self.destroyed) {
+  if (self.destroyed) {
     this.buffer = null;
     this.cb();
     return;
@@ -578,10 +580,7 @@ function processCallback() {
   this.cb();
 }
 
-function _close(engine, callback) {
-  if (callback)
-    process.nextTick(callback);
-
+function _close(engine) {
   // Caller may invoke .close after a zlib error (which will null _handle).
   if (!engine._handle)
     return;
@@ -678,7 +677,7 @@ ObjectSetPrototypeOf(Zlib, ZlibBase);
 function paramsAfterFlushCallback(level, strategy, callback) {
   assert(this._handle, 'zlib binding closed');
   this._handle.params(level, strategy);
-  if (!this._hadError) {
+  if (!this.destroyed) {
     this._level = level;
     this._strategy = strategy;
     if (callback) callback();

test/parallel/test-zlib-destroy.js

@@ -1,13 +1,31 @@
 'use strict';
 
-require('../common');
+const common = require('../common');
 
 const assert = require('assert');
 const zlib = require('zlib');
 
 // Verify that the zlib transform does clean up
 // the handle when calling destroy.
 
-const ts = zlib.createGzip();
-ts.destroy();
-assert.strictEqual(ts._handle, null);
+{
+  const ts = zlib.createGzip();
+  ts.destroy();
+  assert.strictEqual(ts._handle, null);
+
+  ts.on('close', common.mustCall(() => {
+    ts.close(common.mustCall());
+  }));
+}
+
+{
+  // Ensure 'error' is only emitted once.
+  const decompress = zlib.createGunzip(15);
+
+  decompress.on('error', common.mustCall((err) => {
+    decompress.close();
+  }));
+
+  decompress.write('something invalid');
+  decompress.destroy(new Error('asd'));
+}