Observable类型create函数抛出错误未终止问题及优化请求

原代码及问题 #

原代码中，create方法调用异步函数checkUserAndThrowError时未等待其执行完成，导致即便该函数抛出错误，create仍会继续执行后续逻辑：

create(signUpRequestDto: SignUpRequestDto): Observable<Omit<User, 'password'>> {
    this.checkUserAndThrowError(signUpRequestDto.email);
    return this.authService.hashPassword(signUpRequestDto.password).pipe(
      map(hashedPassword => {
        const user = this.userFactory.createUser(signUpRequestDto.email, hashedPassword);
        const { password, ...result } = user;
        this.userRepository.save(user);
        return result as Omit<User, 'password'>;
      }),
    );
  }

async checkUserAndThrowError(email: string): Promise<void> {
    const user = await this.userRepository.findOne({ where: { email } });
    if (user) {
      throw new AuthException(AuthExceptionType.CONFLICT_DUPLICATE_USER);
    }
  }

问题核心：checkUserAndThrowError是异步Promise函数，直接调用不会阻塞create的执行，其抛出的错误属于Promise reject，不会同步终止create的后续逻辑，导致错误无法及时中断流程。

自行编写的代码问题 #

你自己写的代码存在逻辑错误：of(this.userRepository.findOne(...))返回的是Observable包裹的Promise实例，而非实际查询结果，if(user)判断永远为true，会直接抛出错误，完全无法正确校验用户是否存在：

create(signUpRequestDto: SignUpRequestDto): Observable<Omit<User, 'password'>> {

    const user = of(this.userRepository.findOne({ where: { email: signUpRequestDto.email } }));

    if (user) {
      throw new AuthException(AuthExceptionType.CONFLICT_DUPLICATE_USER);
    }
    // this.checkUserAndThrowError(signUpRequestDto.email);
    return this.authService.hashPassword(signUpRequestDto.password).pipe(
      map(hashedPassword => {
        const user = this.userFactory.createUser(signUpRequestDto.email, hashedPassword);
        const { password, ...result } = user;
        this.userRepository.save(user);
        return result as Omit<User, 'password'>;
      }),
    );
  }

正确优化方案 #

需要将异步校验逻辑整合到Observable流中，利用RxJS操作符处理异步流程，确保错误能中断后续操作：

create(signUpRequestDto: SignUpRequestDto): Observable<Omit<User, 'password'>> {
    // 将Promise转为Observable，校验用户是否存在
    return from(this.userRepository.findOne({ where: { email: signUpRequestDto.email } })).pipe(
      // 校验用户存在则抛出错误，中断流
      switchMap(user => {
        if (user) {
          throw new AuthException(AuthExceptionType.CONFLICT_DUPLICATE_USER);
        }
        // 用户不存在则继续执行密码哈希
        return this.authService.hashPassword(signUpRequestDto.password);
      }),
      map(hashedPassword => {
        const user = this.userFactory.createUser(signUpRequestDto.email, hashedPassword);
        const { password, ...result } = user;
        this.userRepository.save(user);
        return result as Omit<User, 'password'>;
      })
    );
  }

优化点说明 #

1. 使用from()将userRepository.findOne()返回的Promise转为Observable，整合到RxJS流中统一处理
2. 通过switchMap操作符串联异步逻辑：若用户存在则抛出错误，直接终止流；不存在则继续执行密码哈希
3. 整个流程在Observable流中执行，错误会自动终止后续操作，无需额外编写中断逻辑

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