[v13.x backport] zlib: align with streams

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, 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', _close.bind(null, this));
  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,29 @@
 '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());
+}
+
+{
+ // 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'));
+}

test/parallel/test-zlib-object-write.js

@@ -5,7 +5,9 @@ const assert = require('assert');
 const { Gunzip } = require('zlib');
 
 const gunzip = new Gunzip({ objectMode: true });
-assert.throws(
- () => gunzip.write({}),
- TypeError
-);
+gunzip.write({}, (err) => {
+ assert.strictEqual(err.code, 'ERR_INVALID_ARG_TYPE');
+});
+gunzip.on('error', (err) => {
+ assert.strictEqual(err.code, 'ERR_INVALID_ARG_TYPE');
+});

test/parallel/test-zlib-write-after-close.js

@@ -27,12 +27,7 @@ const zlib = require('zlib');
 zlib.gzip('hello', common.mustCall(function(err, out) {
  const unzip = zlib.createGunzip();
  unzip.close(common.mustCall());
- assert.throws(
- () => unzip.write(out),
- {
- code: 'ERR_STREAM_DESTROYED',
- name: 'Error',
- message: 'Cannot call write after a stream was destroyed'
- }
- );
+ unzip.write(out, (err) => {
+ assert.strict(err.code, 'ERR_STREAM_DESTROYED');
+ });
 }));