如何用NestJS和RxJS取消除最新请求外的所有旧请求？

原代码核心问题 #

1. 跨请求干扰：控制器是单例，类级别的destroy$被所有请求共享，不同用户的请求会互相取消，完全不符合“仅取消当前用户旧请求”的预期。
2. 异步逻辑冲突：findOne标记为async但返回Observable，NestJS会优先处理Promise，导致RxJS的取消逻辑失效，进而引发异常。
3. 错误捕获不合理：catchError捕获了所有错误（包括takeUntil触发的正常流完成），导致不必要的错误抛出，甚至触发500异常。

方案一：单用户/无用户区分场景（仅处理最新请求） #

如果业务允许所有请求共用“仅处理最新”的逻辑（比如后台单入口操作），直接利用switchMap的自动取消特性即可：

@Controller('user')
export class UserController {
  // 接收所有请求参数的触发源
  private readonly requestTrigger$ = new Subject<{ email: string; username?: string }>();
  // 处理最新请求的流，switchMap自动取消旧请求
  private readonly latestResponse$ = this.requestTrigger$.pipe(
    switchMap(({ email, username }) => 
      of(null).pipe(
        delay(5000), // 模拟长耗时操作
        // 若服务返回Promise，用from转换为Observable
        switchMap(() => from(this.userService.findOne(email, username))),
        // 仅捕获业务逻辑错误，取消操作不会触发此处
        catchError(err => throwError(() => new HttpException(err.message, HttpStatus.BAD_REQUEST)))
      )
    ),
    share() // 共享订阅，避免重复创建流
  );

  constructor(private readonly userService: UserService) {}

  @Get()
  findOne(@Query('email') email: string, @Query('username') username?: string) {
    // 推送新请求，触发switchMap切换到最新任务
    this.requestTrigger$.next({ email, username });
    return this.latestResponse$;
  }
}

方案二：多用户场景（区分客户端，仅取消同一用户旧请求） #

如果需要避免跨用户干扰，通过请求标识（如Session ID）维护每个客户端的独立取消信号：

@Controller('user')
export class UserController {
  // 存储每个客户端的取消信号
  private readonly clientCancelMap = new Map<string, Subject<void>>();

  constructor(private readonly userService: UserService) {}

  @Get()
  findOne(
    @Query('email') email: string,
    @Query('username') username?: string,
    @Req() req: Request // 需导入express的Request类型
  ) {
    // 用Session ID作为客户端唯一标识（需提前配置NestJS Session中间件）
    const clientId = req.session.id;
    
    // 取消当前客户端的旧请求
    const oldCancel$ = this.clientCancelMap.get(clientId);
    if (oldCancel$) {
      oldCancel$.next();
      oldCancel$.complete();
      this.clientCancelMap.delete(clientId);
    }

    // 创建新的取消信号
    const newCancel$ = new Subject<void>();
    this.clientCancelMap.set(clientId, newCancel$);

    return of(null).pipe(
      delay(5000),
      switchMap(() => from(this.userService.findOne(email, username))),
      takeUntil(newCancel$), // 绑定当前客户端的取消信号
      // 请求完成/取消后清理资源
      finalize(() => {
        newCancel$.complete();
        this.clientCancelMap.delete(clientId);
      }),
      catchError(err => throwError(() => new HttpException(err.message, HttpStatus.BAD_REQUEST)))
    );
  }
}

关键注意事项 #

• 若userService.findOne返回Promise，必须用from()转换为Observable，才能融入RxJS流。
• 不要滥用catchError，仅捕获业务逻辑错误，takeUntil触发的流完成属于正常操作，无需捕获。
• 方案二中的Session ID需要提前配置NestJS的Session中间件（如express-session），也可以用Token、IP+UserAgent组合作为客户端标识。

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