如何等待流完成管道传输？ （Nodejs）

Question

我有一个 for 循环的 Promise 数组，所以我使用 Promise.all 来遍历它们，然后调用 then 。

let promises = [];
promises.push(promise1);
promises.push(promise2);
promises.push(promise3);

Promise.all(promises).then((responses) => {
  for (let i = 0; i < promises.length; i++) {
    if (promise.property === something) {
      //do something
    } else {
      let file = fs.createWriteStream('./hello.pdf');
      let stream = responses[i].pipe(file);
      /*
         I WANT THE PIPING AND THE FOLLOWING CODE 
         TO RUN BEFORE NEXT ITERATION OF FOR LOOP
      */
      stream.on('finish', () => {
        //extract the text out of the pdf
        extract(filePath, {splitPages: false}, (err, text) => {
        if (err) {
          console.log(err);
        } else {
          arrayOfDocuments[i].text_contents = text;
        }
      });
    });    
  }
}

promise1、promise2 和promise3 是一些http 请求，如果其中一个是应用程序/pdf，那么我将其写入流并解析其中的文本。但这段代码在解析 pdf 中的测试之前运行下一次迭代。有没有办法让代码等到流的管道和提取完成后再继续下一次迭代？

Answer 1

如果没有 async/await，那是相当令人讨厌的。使用 async/await，只需执行以下操作：

Promise.all(promises).then(async (responses) => {
  for (...) {
    await new Promise(resolve => stream.on("finish", resolve));
    //extract the text out of the PDF
  }
})

Answer 2

像下面这样的东西也可以工作。我经常使用这种模式：

let promises = [];
promises.push(promise1);
promises.push(promise2);
promises.push(promise3);

function doNext(){
  if(!promises.length) return;
  promises.shift().then((resolved) =>{
    if(resolved.property === something){
      ...
      doNext();
    }else{
      let file = fs.createWriteStream('./hello.pdf');
      let stream = resolved.pipe(file);
      stream.on('finish', () =>{
        ...
        doNext();
      });
    }

  })
}
doNext();

或者将处理程序分解为控制器和 Promisified 处理程序：

function streamOrNot(obj){
  return new Promise(resolve, reject){
    if(obj.property === something){
      resolve();
      return;
    }
    let file = fs.createWriteStream...;
    stream.on('finish', () =>{
      ...
      resolve();
    });
  }
}

function doNext(){
  if(!promises.length) return;
  return promises.shift().then(streamOrNot).then(doNext);
}

doNext()

Answer 3

将

await

与

stream.pipeline()

一起使用，而不是

stream.pipe()

：

import * as StreamPromises from "stream/promises";

...
await StreamPromises.pipeline(sourceStream, destinationStream);

Answer 4

您可以将 else 部分写在自调用函数中。这样流的处理就会并行发生

(function(i) {
    let file = fs.createWriteStream('./hello.pdf');
    let stream = responses[i].pipe(file);
  /*
     I WANT THE PIPING AND THE FOLLOWING CODE 
     TO RUN BEFORE NEXT ITERATION OF FOR LOOP
  */
    stream.on('finish', () => {
      //extract the text out of the pdf
      extract(filePath, {splitPages: false}, (err, text) => {
      if (err) {
        console.log(err);
      } 
      else {
        arrayOfDocuments[i].text_contents = text;
      }
    });
  });    
})(i) 

否则，您可以将流媒体部分作为原始/个人承诺本身的一部分来处理。

到目前为止，您正在创建 Promise 并将其添加到数组中，而不是将 Promise.then 添加到数组中（这也是一个 Promise）。然后在处理程序内部进行流处理。

Answer 5

我相信在将回调附加到

finish

事件后，您可以简单地为 Node.js 15+ 执行此操作...

const { finished } = require('node:stream/promises');

await finished(stream);

https://nodejs.org/api/stream.html#streamfinishedstream-options