node.js - Node : how can I wait for pipe and createWriteStream to finish before executing next lines?

Question

我是 Nodejs 新手。我正在将一个 zip 文件从 S3 存储桶保存到托管 nodejs 服务的 EC2 实例，然后我在 EC2 的文件系统中本地解压缩该 zip 文件:

  const s3Item = await S3.getFile(objectBucket, objectKey);
  let stream = s3Item.Body.pipe(fs.createWriteStream(sourceFilePath, { mode: 0o777 }));
  stream.on('finish', () => {
    logger.info(` pipe done `);  
  });


  logger.info(`start decompressing...`);
  await decompressArchive(sourceFilePath, targetDirectory_decompressed);

  ... a lot of code...

但是，start decompressing... 总是在 pipe done 打印之前打印。

我怎样才能使这两个步骤同步，以便我们等到 pipe done，然后开始解压缩？

我想使用 async/await ，因为我根本无法将所有其余代码放在 stream.on('finish', () => {}); block 中，因为代码太多了，都依赖于正在完成的流。

我已经搜索了相关的答案(有很多)，但我仍然无法正常工作。

Answer 1

好吧，流是事件驱动的。您可以将流完成后要运行的代码放入 finish 事件处理程序中:

  const s3Item = await S3.getFile(objectBucket, objectKey);
  let stream = s3Item.Body.pipe(fs.createWriteStream(sourceFilePath, { mode: 0o777 }));
  stream.on('finish', async () => {
    logger.info(` pipe done `);  
    logger.info(`start decompressing...`);
    await decompressArchive(sourceFilePath, targetDirectory_decompressed);
  });

或者，您可以将 finish 事件包装在一个 promise 和 await 中:

  const s3Item = await S3.getFile(objectBucket, objectKey);
  let stream = s3Item.Body.pipe(fs.createWriteStream(sourceFilePath, { mode: 0o777 }));

  await new Promise((resolve, reject) => {
    stream.on('finish', () => {
      logger.info(` pipe done `);
      resolve();  
    }).on('error', err => {
      reject(err);
    });
  });

  logger.info(`start decompressing...`);
  await decompressArchive(sourceFilePath, targetDirectory_decompressed);

仅供引用，最新版本的 nodejs 有一个 once()事件模块中的函数使这更容易一些:

  const { once } = require('events');

  const s3Item = await S3.getFile(objectBucket, objectKey);
  let stream = s3Item.Body.pipe(fs.createWriteStream(sourceFilePath, { mode: 0o777 }));

  await once(stream, 'finish');
  logger.info(` pipe done `);

  logger.info(`start decompressing...`);
  await decompressArchive(sourceFilePath, targetDirectory_decompressed);

或者，您可以使用 pipeline() 的 promisified 版本代替 .pipe()。有很多方法可以做到这一点。