node.js - 了解 Node.js 中的流

Question

我遇到了一个问题，我需要使用 NodeJs 创建和下载多个文件的 zip。我尝试过但失败的事情:

https://github.com/archiverjs/node-archiver/issues/364#issuecomment-573508251

https://github.com/kubernetes/kubernetes/issues/90483#issue-606722433

unexpected behavior using zip-stream NPM on Google k8s

除此之外，现在文件也已加密，因此我必须在将它们动态添加到 zip 之前对其进行解密。

虽然我也解决了这个问题，但我的解决方案在服务器在本地机器上运行时运行良好，但在 Google Kubernetes Engine 上运行时失败。

经过更多研究，我猜这可能是因为 NodeJs 中的流中存在背压问题，但如文档中所述，背压由 pipe 方法自动处理。有没有可能是浏览器的接收速度和我的服务器的发送速度不匹配/如果是压缩如何解决这个问题？

与问题相关的所有示例都在上面提供的链接中。

除此可读流外，还通过解密器对其进行解密。

    const handleEntries = ({ elem, uniqueFiles, archive, speedLimit }) => {
  return new Promise((resolve, reject) => {
    let fileName = elem.fileName;
    const url = elem.url;
    const decipher = elem.decipher;
    // changing fileName if same filename is already added to zip
    if (uniqueFiles[fileName] || uniqueFiles[fileName] === 0) {
      uniqueFiles[fileName]++;
    } else {
      uniqueFiles[fileName] = 0;
    }
    if (uniqueFiles[fileName]) {
      const lastDotIndex = fileName.lastIndexOf(".");
      const name = fileName.substring(0, lastDotIndex);
      const extension = fileName.substring(lastDotIndex + 1);
      fileName = `${name}(${uniqueFiles[fileName]}).${extension}`;
    }
    let readableStream = Request(url);
    // create a "Throttle" instance that reads at speedLimit bps
    if (speedLimit) {
      const throttle = new Throttle({ bps: Number(speedLimit) });
      readableStream = readableStream.pipe(throttle);
    }
    // if file is encrypted, need to decrypt it before piping to zip
    readableStream = decipher ? readableStream.pipe(decipher) : readableStream;
    archive.append(readableStream, { name: fileName });
    readableStream.on("complete", result => {
      console.log("Request stream event complete : ", fileName);
      resolve("done");
      // readableStream.unpipe();
      // readableStream.destroy();
    });
    readableStream
      .on("error", error => {
        console.log("Request stream event error fileName : ", fileName, " error : ", error);
        // readableStream.unpipe();
        // readableStream.destroy();
        resolve("done");
      })
      .on("pipe", result => {
        console.log("Request stream event pipe : ", fileName);
      })
      .on("request", result => {
        console.log("Request stream event request : ", fileName);
      })
      .on("response", result => {
        console.log("Request stream event response : ", fileName);
      })
      .on("socket", result => {
        result.setKeepAlive(true);
        console.log("Request stream event socket : ", fileName);
      });
  });
};

const useArchiver = async ({ resp, urls, speedLimit }) => {
  resp.writeHead(200, {
    "Content-Type": "application/zip",
    "Content-Disposition": `attachment; filename="${outputFileName}"`,
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, POST, OPTIONS"
  });
  const uniqueFiles = {};
  const archive = Archiver("zip", { zlib: 0 });
  archive.pipe(resp);
  archive
    .on("close", result => {
      console.log("archive stream event close : ", result);
      // archive.unpipe();
      // archive.destroy();
    })
    .on("drain", result => {
      console.log("archive stream event drain : ", result);
    })
    .on("entry", result => {
      console.log("archive stream event entry : ", result.stats);
    })
    .on("error", error => {
      console.log("archive stream event error : ", error);
      reject("error");
      // archive.unpipe();
      // archive.destroy();
    })
    .on("finish", result => {
      console.log("archive stream event finish : ", result);
      // archive.unpipe();
      // archive.destroy();
    })
    .on("pipe", result => {
      console.log("archive stream event pipe : ");
    })
    .on("progress", async result => {
      console.log("archive stream event progress : ", result.entries);
      if (urls.length === result.entries.total && urls.length === result.entries.processed) {
        await archive.finalize();
        console.log("finalized : ", urls[0]);
      }
    })
    .on("unpipe", result => {
      console.log("archive stream event unpipe : ");
    })
    .on("warning", result => {
      console.log("archive stream event warning : ", result);
    });
  for (const elem of urls) {
    await handleEntries({ elem, uniqueFiles, archive, speedLimit });
  }
};

我用归档器尝试了这段代码，在压缩大文件时获取归档器的耗尽事件，管道是否处理背压，如果是，为什么我从归档器获取耗尽事件。

Answer 1

嘿)我研究过你的代码，可以告诉你，你使用了一个 promise 函数并且你没有等到她完成。您需要使用 await new Promise() 包装 zipStreamer.entry。而且一定是这样的

async function doSmth() {
   const decipher = crypto.createDecipheriv(
      algorithm,
      Buffer.from(key), 
      Buffer.from(key.substring(0, 9)
   ));
   await new Promise((resolve, reject) => {
     zipStreamer.entry(readableStream.pipe(decipher), {
      name: fileName
     }, (error, result) => {
      if (!error) {
         resolve("done");
      } else {
        reject("error");
      }
     });  
   });
}