Node.Js 流管道上的异步迭代器

Question

我有以下管道:

readFile > parseCSV > otherProcess

readFile 是标准 Node.Js createReadStream，而 parseCSV 是 Node.js 转换流(模块 link)。

我想逐行迭代 csv 文件并一次处理一行。因此，流和异步迭代器是完美的搭配。

我有以下代码可以正常工作:

async function* readByLine(path, opt) {
  const readFileStream = fs.createReadStream(path);
  const csvParser = parse(opt);
  const parser = readFileStream.pipe(csvParser);
  for await (const record of parser) {
    yield record;
  }
}

我对 Node.Js 流还很陌生，但我从许多来源了解到，模块 stream.pipeline 优于 .pipe 方法读取流。

如何更改上面的代码以使用 stream.pipeline (实际上是从 util.promisify(pipeline) 获得的 promise 版本)并产生一行当时？

Answer 1

添加到@eol的答案中，我建议存储 promise 并在异步迭代后等待它。

const fs = require('fs');
const parse = require('csv-parse');
const stream = require('stream');

async function* readByLine(path, opt) {
    const readFileStream = fs.createReadStream(path);
    const csvParser = parse(opt);
    const promise = stream.promises.pipeline(readFileStream, csvParser);
    for await (const record of csvParser) {
        yield record;
    }
    await promise;
}

通过在循环之前调用 await pipeline(...)，它将消耗整个流，然后您才能从缓冲区中剩余的内容进行迭代，这在小流上偶然起作用，但可能会在较大(或无限/惰性)流上中断。

根据我们等待的位置，回调等效项可能会更清楚地表明发生了什么。

// await before iterating
stream.pipeline(a, b, err => {
  if (err) return callback(err)

  for await (const record of b) {
    // process record
  }

  callback()
}

// await after iterating
for await (const record of stream.pipeline(a, b, callback)) {
  // process record
}