如何在NodeJS流中处理错误并返回NestJS HTTP错误响应？

流的data事件属于异步回调，抛出的异常不在Nest请求处理的上下文内，所以Nest的异常过滤器无法捕获，只会触发Node的未捕获异常警告。以下是几种可行的解决思路：

1. 用Promise包装流处理逻辑 #

将整个流操作包装成Promise，在验证失败或流出错时reject异常，控制器通过async/await捕获后抛出，让Nest的异常过滤器处理并返回响应：

import { UnprocessableEntityException } from '@nestjs/common';
import { createReadStream } from 'fs';
import csv from 'csv-parser';

async handleFileUpload(@UploadedFile() file: Express.Multer.File) {
  const spreadsheet = [];

  return new Promise((resolve, reject) => {
    const stream = createReadStream(file.path);

    stream
      .pipe(csv())
      .on('data', (data) => {
        try {
          spreadsheetValidationSchema.parse(data);
          spreadsheet.push(data);
        } catch (err) {
          // 验证失败时销毁流，避免内存泄漏
          stream.destroy();
          reject(new UnprocessableEntityException('数据格式不符合要求'));
        }
      })
      .on('end', () => {
        resolve(spreadsheet);
      })
      .on('error', (streamErr) => {
        reject(new UnprocessableEntityException('文件流处理失败'));
      });
  });
}

2. 使用stream.pipeline（推荐） #

Node.js的stream/promises模块提供的pipeline方法会自动处理流的销毁、错误捕获，更安全可靠，适合异步场景：

import { UnprocessableEntityException } from '@nestjs/common';
import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';
import csv from 'csv-parser';
import { ZodError } from 'zod';

async handleFileUpload(@UploadedFile() file: Express.Multer.File) {
  const spreadsheet = [];

  try {
    await pipeline(
      createReadStream(file.path),
      csv(),
      // 用异步迭代器处理每一行数据
      async function* (source) {
        for await (const data of source) {
          spreadsheetValidationSchema.parse(data);
          spreadsheet.push(data);
        }
      }
    );
    return spreadsheet;
  } catch (err) {
    // 区分验证错误和流错误，返回更精准的提示
    if (err instanceof ZodError) {
      throw new UnprocessableEntityException(`数据验证失败：${err.message}`);
    }
    throw new UnprocessableEntityException('文件处理出错');
  }
}

核心原理 #

流的异步回调脱离了Nest请求处理的上下文，直接抛出异常无法被Nest的异常捕获机制拦截。通过Promise包装或使用pipeline的Promise版本，将异常重新纳入请求处理的异步链中，就能让Nest正常捕获并返回对应的HTTP错误响应。

内容的提问来源于stack exchange，提问作者Timóteo Stifft