[minor] Refactor PerMessageDeflate#_decompress()

If context takeover is enabled there is no reason to remove and add
the listeners of the `'data'` and `'error'` events every time
`PerMessageDeflate#_decompress()` is used.

lib/PerMessageDeflate.js

@@ -14,7 +14,10 @@ const EMPTY_BLOCK = Buffer.from([0x00]);
 const kWriteInProgress = Symbol('write-in-progress');
 const kPendingClose = Symbol('pending-close');
 const kTotalLength = Symbol('total-length');
+const kCallback = Symbol('callback');
 const kBuffers = Symbol('buffers');
+const kError = Symbol('error');
+const kOwner = Symbol('owner');
 
 // We limit zlib concurrency, which prevents severe memory fragmentation
 // as documented in https://github.com/nodejs/node/issues/8871#issuecomment-250915913
@@ -321,53 +324,47 @@ class PerMessageDeflate {
         : this.params[key];
 
       this._inflate = zlib.createInflateRaw({ windowBits });
+      this._inflate[kTotalLength] = 0;
+      this._inflate[kBuffers] = [];
+      this._inflate[kOwner] = this;
+      this._inflate.on('error', inflateOnError);
+      this._inflate.on('data', inflateOnData);
     }
-    this._inflate[kWriteInProgress] = true;
-
-    var totalLength = 0;
-    const buffers = [];
-    var err;
 
-    const onData = (data) => {
-      totalLength += data.length;
-      if (this._maxPayload < 1 || totalLength <= this._maxPayload) {
-        return buffers.push(data);
-      }
+    this._inflate[kCallback] = callback;
+    this._inflate[kWriteInProgress] = true;
 
-      err = new Error('max payload size exceeded');
-      err.closeCode = 1009;
-      this._inflate.reset();
-    };
+    this._inflate.write(data);
+    if (fin) this._inflate.write(TRAILER);
 
-    const onError = (err) => {
-      cleanup();
-      callback(err);
-    };
+    this._inflate.flush(() => {
+      const err = this._inflate[kError];
 
-    const cleanup = () => {
-      if (!this._inflate) return;
+      if (err) {
+        this._inflate.close();
+        this._inflate = null;
+        callback(err);
+        return;
+      }
 
-      this._inflate.removeListener('error', onError);
-      this._inflate.removeListener('data', onData);
-      this._inflate[kWriteInProgress] = false;
+      const data = bufferUtil.concat(
+        this._inflate[kBuffers],
+        this._inflate[kTotalLength]
+      );
 
       if (
         (fin && this.params[`${endpoint}_no_context_takeover`]) ||
         this._inflate[kPendingClose]
       ) {
         this._inflate.close();
         this._inflate = null;
+      } else {
+        this._inflate[kWriteInProgress] = false;
+        this._inflate[kTotalLength] = 0;
+        this._inflate[kBuffers] = [];
       }
-    };
 
-    this._inflate.on('error', onError).on('data', onData);
-    this._inflate.write(data);
-    if (fin) this._inflate.write(TRAILER);
-
-    this._inflate.flush(() => {
-      cleanup();
-      if (err) callback(err);
-      else callback(null, bufferUtil.concat(buffers, totalLength));
+      callback(null, data);
     });
   }
 
@@ -419,6 +416,7 @@ class PerMessageDeflate {
         this._deflate[kBuffers],
         this._deflate[kTotalLength]
       );
+
       if (fin) data = data.slice(0, data.length - 4);
 
       if (
@@ -450,3 +448,41 @@ function deflateOnData (chunk) {
   this[kBuffers].push(chunk);
   this[kTotalLength] += chunk.length;
 }
+
+/**
+ * The listener of the `zlib.InflateRaw` stream `'data'` event.
+ *
+ * @param {Buffer} chunk A chunk of data
+ * @private
+ */
+function inflateOnData (chunk) {
+  this[kTotalLength] += chunk.length;
+
+  if (
+    this[kOwner]._maxPayload < 1 ||
+    this[kTotalLength] <= this[kOwner]._maxPayload
+  ) {
+    this[kBuffers].push(chunk);
+    return;
+  }
+
+  this[kError] = new Error('max payload size exceeded');
+  this[kError].closeCode = 1009;
+  this.removeListener('data', inflateOnData);
+  this.reset();
+}
+
+/**
+ * The listener of the `zlib.InflateRaw` stream `'error'` event.
+ *
+ * @param {Error} err The emitted error
+ * @private
+ */
+function inflateOnError (err) {
+  //
+  // There is no need to call `Zlib#close()` as the handle is automatically
+  // closed when an error is emitted.
+  //
+  this[kOwner]._inflate = null;
+  this[kCallback](err);
+}