如何将Node.js stream错误事件传播到async/await风格代码的捕获块

你当前写法的问题在于，pipe()方法不会自动在流管道链上传递错误，每个流的error事件只会触发自身的监听器。而for await...of迭代流时，仅会捕获你所迭代的目标流（也就是你返回的transform2）自身抛出的错误，上游的readable、csvParser、transform1的错误无法自动传递到transform2，因此不会被外层catch捕获。

方案1：手动转发错误到最终输出流 #

把所有上游流的错误都转发给你最终返回的transform2流，这样所有错误都会被for await感知到：

function getStream() {
  // transform1和transform2为自定义转换流
  const finalStream = readable.pipe(csvParser).pipe(transform1).pipe(transform2);

  // 遍历所有流，统一转发错误
  [readable, csvParser, transform1, transform2].forEach(stream => {
    stream.on('error', err => {
      // 可在此处添加你的错误日志逻辑
      // 转发错误到最终输出流
      finalStream.destroy(err);
    });
  });

  return finalStream;
}

async function main() {
  try {
    const stream = getStream();
    for await(const chunk of stream) {
      // 处理chunk
    }
  } catch (ex) {
    // 所有流的错误都会被捕获到这里
    console.log('捕获到流错误：', ex);
  }
}

方案2：使用官方推荐的stream.pipeline（更安全） #

Node.js内置的stream.pipeline会自动处理管道错误传播、异常时的流资源销毁，避免内存泄漏，配合Promise版本的pipeline可以直接捕获所有错误：

// 引入Promise版本的pipeline
const { pipeline } = require('stream/promises');

async function main() {
  try {
    await pipeline(
      readable,
      csvParser,
      transform1,
      transform2,
      async function* (source) {
        for await (const chunk of source) {
          // 此处放置你原来的chunk处理逻辑
          // 处理完成后如果不需要继续输出可以不用yield
          yield chunk;
        }
      }
    );
  } catch (ex) {
    // 所有管道环节的错误都会被捕获到这里
    console.log('捕获到流错误：', ex);
  }
}

内容的提问来源于stack exchange，提问作者krl