NestJS 拦截器HTTP正常但WebSocket场景下数据修改不生效问题

问题描述 #

开发SignFileInterceptor拦截器，用于在控制器/网关处理逻辑执行完成后修改返回数据：

• HTTP场景下通过APP_INTERCEPTOR令牌全局注册拦截器，所有请求拦截正常，数据修改结果符合预期
• WS场景下通过@UseInterceptors(SignFileInterceptor)装饰器在网关类上引入拦截器后，拦截器代码会执行，但内部异步逻辑未执行完成时原始数据就会被发送给客户端，最终客户端收到未修改的原始数据

现有实现代码 #

拦截器实现 #

@Injectable()
export class SignFileInterceptor implements NestInterceptor {
  constructor(private fileService: FilesService) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(
      map(async (data) => {
        const paths = getFilesDtoPaths(data);

        for (const path of paths) {
          const file = get(data, path);

          // 返回Promise类型结果
          const signed = await this.fileService.signFile(file);

          set(data, path, signed);
        }

        return data;
      }),
    );
  }
}

HTTP场景全局注册配置 #

providers: [
    AppService,
    {
      provide: APP_INTERCEPTOR,
      useClass: SignFileInterceptor,
    }
 ]

WS网关引入拦截器配置 #

@WebSocketGateway({
  cors,
  allowEIO3: true,
})
@UseInterceptors(SignFileInterceptor)
@Injectable()
export class EventsGateway {
  constructor() {}

  @WebSocketServer()
  server!: Server;

  @SubscribeMessage('name1')
  async handleJoinRoom(): Promise<string> {
    return 'john doe'
  }
  
  @SubscribeMessage('name2')
  async handleJoinRoom(): Promise<string> {
    return 'john doe 2'
  }
}

问题根因 #

核心出在拦截器里RxJS操作符的使用错误：

• map操作符不会等待传入的async函数执行完成，会直接把async返回的Promise对象作为流的值传递给后续处理流程
• Nest的HTTP适配器内置了Promise值解析逻辑，会等待Promise resolve后再发送响应，因此HTTP场景下表现正常
• Nest的WS适配器默认不会解析流中嵌套的Promise值，拿到值后会直接序列化为消息发送给客户端，因此会出现异步逻辑未执行完、原始数据提前发出的问题

修复方案 #

将拦截器中的map操作符替换为mergeMap，该操作符会自动等待内部异步逻辑（Promise/Observable）执行完成，再将最终处理结果传递给后续流程。修改后的拦截器可同时兼容HTTP、WS两个场景，无需拆分逻辑。

修改后的拦截器代码：

import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
import { Observable } from 'rxjs';
// 注意导入mergeMap
import { mergeMap } from 'rxjs/operators';

@Injectable()
export class SignFileInterceptor implements NestInterceptor {
  constructor(private fileService: FilesService) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(
      mergeMap(async (data) => {
        const paths = getFilesDtoPaths(data);

        for (const path of paths) {
          const file = get(data, path);
          const signed = await this.fileService.signFile(file);
          set(data, path, signed);
        }

        return data;
      }),
    );
  }
}

补充说明 #

• 如果需要严格保证异步逻辑的执行顺序，可将mergeMap替换为concatMap，效果一致
• 修改后原有HTTP场景的拦截逻辑不受影响，不需要调整全局注册配置

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