使用NestJS+fast-csv监听流事件时返回空数据的问题

在使用NestJS结合fast-csv库计算导出文件流的大小时遇到异常：原本代码可正常导出CSV文件，但启用stream的data事件监听器计算流大小后，虽能获取到流大小数值，但接口返回的CSV文件内容为空。

服务代码 #

import { format } from 'fast-csv';

async export() {
    const stream = format({ headers: true });

    stream.write(['id', 'first_name', 'last_name', 'is_active']);

    let size = 0;

    // 启用这段代码后出现问题
    // stream.on('data', (chunk) => {
    //   size += chunk.length;
    // });

    // stream.on('end', () => {
    //   console.log(size);
    // });

    await this.getData(stream)().finally(() => {
      stream.end();
    });

    return stream ? { stream, size } : null;
  }

  private getData = (stream: Transform) => async () => {
    for (let i = 0; i < 1; i++) {
      const users = await User.find();

      if (users.length === 0) {
        break;
      }

      for (let i = 0; i < users.length; i++) {
        const row = users[i];

        stream.write([row.id, row.firstName, row.lastName, row.isActive]);
      }
    }
  };

控制器代码 #

@Post('export')
  async exporrt() {
    const data = await this.appService.export();

    console.log('size', data.size);

    return new StreamableFile(data.stream, {
      disposition: `attachment; filename=test.csv`,
      type: 'text/csv; charset=utf-8',
    });
  }

问题根源 #

Node.js的流在绑定data事件后会自动切换到流动模式，如果只监听data事件却不消费（将数据传递给下游），流中的chunk会直接被丢弃，最终返回给客户端的流自然为空。

方法一：通过PassThrough中转流实现统计 #

创建一个PassThrough流作为中转，让fast-csv的输出先流经这个中转流，在中转流上监听data事件计算大小，同时将中转流返回给客户端，既保证数据不丢失，又能完成大小统计。

修改后的服务代码：

import { format } from 'fast-csv';
import { PassThrough } from 'stream';

async export() {
    const csvStream = format({ headers: true });
    const passThrough = new PassThrough(); // 中转流
    let size = 0;

    // 在中转流上监听data事件累加大小
    passThrough.on('data', (chunk) => {
      size += chunk.length;
    });

    passThrough.on('end', () => {
      console.log('文件总字节数:', size);
    });

    // 将csv流的数据导入中转流
    csvStream.pipe(passThrough);

    csvStream.write(['id', 'first_name', 'last_name', 'is_active']);

    await this.getData(csvStream)().finally(() => {
      csvStream.end();
    });

    // 返回中转流和统计的大小
    return { stream: passThrough, size };
  }

  // getData方法保持原有逻辑不变
  private getData = (stream: Transform) => async () => {
    for (let i = 0; i < 1; i++) {
      const users = await User.find();

      if (users.length === 0) {
        break;
      }

      for (let i = 0; i < users.length; i++) {
        const row = users[i];

        stream.write([row.id, row.firstName, row.lastName, row.isActive]);
      }
    }
  };

方法二：自定义Transform流封装统计逻辑 #

通过自定义Transform流，在处理每个chunk时自动累加大小，同时将chunk传递给下游，这种方式更符合Node.js流的处理规范。

修改后的服务代码：

import { format } from 'fast-csv';
import { Transform } from 'stream';

async export() {
    const csvStream = format({ headers: true });
    let size = 0;

    // 自定义Transform流，统计大小同时传递数据
    const sizeCounter = new Transform({
      transform(chunk, encoding, callback) {
        size += chunk.length;
        callback(null, chunk); // 将chunk传递给下游流
      }
    });

    sizeCounter.on('end', () => {
      console.log('文件总字节数:', size);
    });

    csvStream.pipe(sizeCounter);

    csvStream.write(['id', 'first_name', 'last_name', 'is_active']);

    await this.getData(csvStream)().finally(() => {
      csvStream.end();
    });

    return { stream: sizeCounter, size };
  }

注意事项 #

• 禁止直接在返回给客户端的流上监听data事件却不消费数据，这是导致数据丢失的核心原因。
• 统计得到的size是字节数，若需要转换为KB/MB等可读单位，可自行添加单位转换逻辑。

内容的提问来源于stack exchange，提问作者Đoàn Đức Bảo