fix: impl _final method instead hack pipe event

Author: fengmk2

.gitignore

@@ -7,3 +7,4 @@ test/fixtures/chinese-path-test.zip
 .DS_Store
 yarn.lock
 !test/fixtures/symlink/node_modules
+pnpm-lock.yaml

lib/zip/uncompress_stream.js

@@ -2,6 +2,7 @@
 
 // https://github.com/thejoshwolfe/yauzl#no-streaming-unzip-api
 
+const debug = require('util').debuglog('compressing/zip/uncompress_stream');
 const yauzl = require('@eggjs/yauzl');
 const stream = require('stream');
 const UncompressBaseStream = require('../base_write_stream');
@@ -38,12 +39,20 @@ class ZipUncompressStream extends UncompressBaseStream {
     if (this._zipFileNameEncoding === 'utf-8') {
       this._zipFileNameEncoding = 'utf8';
     }
+    this._finalCallback = err => {
+      if (err) {
+        debug('finalCallback, error: %j', err);
+        return this.emit('error', err);
+      }
+      this.emit('finish');
+    };
 
     this[YAUZL_CALLBACK] = this[YAUZL_CALLBACK].bind(this);
 
     const sourceType = utils.sourceType(opts.source);
 
     const yauzlOpts = this._yauzlOpts = Object.assign({}, DEFAULTS, opts.yauzl);
+    debug('sourceType: %s, yauzlOpts: %j', sourceType, yauzlOpts);
     if (sourceType === 'file') {
       yauzl.open(opts.source, yauzlOpts, this[YAUZL_CALLBACK]);
       return;
@@ -60,27 +69,27 @@ class ZipUncompressStream extends UncompressBaseStream {
         .catch(e => this.emit('error', e));
       return;
     }
-
-    this.on('pipe', srcStream => {
-      srcStream.unpipe(srcStream);
-
-      utils.streamToBuffer(srcStream)
-        .then(buf => {
-          this._chunks.push(buf);
-          buf = Buffer.concat(this._chunks);
-          yauzl.fromBuffer(buf, yauzlOpts, this[YAUZL_CALLBACK]);
-        })
-        .catch(e => this.emit('error', e));
-    });
   }
 
-  _write(chunk) {
+  _write(chunk, _encoding, callback) {
     // push to _chunks array, this will only happen once, for stream will be unpiped.
     this._chunks.push(chunk);
+    debug('write size: %d, chunks: %d', chunk.length, this._chunks.length);
+    callback();
+  }
+
+  _final(callback) {
+    const buf = Buffer.concat(this._chunks);
+    debug('final, buf size: %d, chunks: %d', buf.length, this._chunks.length);
+    this._finalCallback = callback;
+    yauzl.fromBuffer(buf, this._yauzlOpts, this[YAUZL_CALLBACK]);
   }
 
   [YAUZL_CALLBACK](err, zipFile) {
-    if (err) return this.emit('error', err);
+    if (err) {
+      debug('yauzl error', err);
+      return this._finalCallback(err);
+    }
 
     zipFile.readEntry();
 
@@ -106,17 +115,22 @@ class ZipUncompressStream extends UncompressBaseStream {
 
         if (type === 'file') {
           zipFile.openReadStream(entry, (err, readStream) => {
-            if (err) return this.emit('error', err);
+            if (err) {
+              debug('file, error: %j', err);
+              return this._finalCallback(err);
+            }
+            debug('file, header: %j', header);
             this.emit('entry', header, readStream, next);
           });
         } else { // directory
           const placeholder = new stream.Readable({ read() {} });
+          debug('directory, header: %j', header);
           this.emit('entry', header, placeholder, next);
           setImmediate(() => placeholder.emit('end'));
         }
       })
-      .on('end', () => this.emit('finish'))
-      .on('error', err => this.emit('error', err));
+      .on('end', () => this._finalCallback())
+      .on('error', err => this._finalCallback(err));
 
     function next() {
       zipFile.readEntry();

test/zip/uncompress_stream.test.js

@@ -21,6 +21,7 @@ describe('test/zip/uncompress_stream.test.js', () => {
 
     const uncompressStream = new compressing.zip.UncompressStream();
     fs.mkdirSync(destDir, { recursive: true });
+
     pump(sourceStream, uncompressStream, err => {
       assert(!err);
       const res = dircompare.compareSync(originalDir, path.join(destDir, 'xxx'));