如何将Node.js的fs.createReadStream封装为Promise，解决end事件提前触发问题

问题根因 #

你遇到的end回调提前触发问题，核心是Node.js流的事件系统不会等待异步回调执行完成：你给data事件绑定的async函数返回的Promise会被事件系统直接忽略，流只负责按顺序派发事件，不会等待回调内部的异步逻辑执行完。即便你调用了stream.pause()，最后一次data事件派发时流已经读完了所有数据，此时end事件会同步触发，不会等你当前正在执行的await row(data)逻辑跑完。

1. 如何让end回调等待所有处理逻辑完成 #

你可以新增两个状态标记，分别记录当前是否有正在执行的异步row任务、以及流是否已经触发end事件，等最后一个row任务执行完成后再触发resolve即可，修改后代码如下：

import * as fs from 'fs';

function parseFile<T>(filePath: string, row: (x: T) => Promise<void>) {
  return new Promise((resolve, reject) => {
    const stream = fs.createReadStream(filePath);
    // 记录当前是否有正在处理的row任务
    let processing = false;
    // 标记流是否已经触发end事件
    let endReceived = false;

    stream.on('data', async (data) => {
      processing = true;
      try {
        stream.pause();
        await row(data as T);
      } catch (e) {
        reject(e);
      } finally {
        processing = false;
        stream.resume();
        // 处理完当前任务后如果已经收到end事件，直接resolve
        if (endReceived) resolve();
      }
    })
    .on('end', () => {
      endReceived = true;
      // 如果当前没有正在处理的任务直接resolve，否则等最后一个任务处理完再resolve
      if (!processing) resolve();
    })
    .on('error', (err) => {
      reject(err);
    });
  });
}

注意传入的row函数必须是返回Promise的异步函数，否则await不会生效。修改后你原有的测试用例就会输出预期的1000而非1。

2. close事件未触发的原因 #

fs.createReadStream的close事件是在文件描述符被完全关闭后才会触发，你当前代码未触发该事件，通常是因为流触发end后还有未执行完的异步逻辑持有相关资源引用。就算close事件能正常触发，它和end事件一样是流层面的完成通知，依然不会等待你自定义的异步row任务执行完成，所以无法直接用它解决提前resolve的问题。

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