javascript - 嵌套的可观察对象仅在超时后传播错误

Question

我的组件中有一个方法，它使用 observable 来监听第一个事件，并且超时时间为 15 秒。该方法调用一个嵌套的可观察量，如果参数为 null 或空，则该可观察量应抛出错误。我使用 Observable.throw() 但错误仅在超时/15 秒后传播。

this.signUp(this.user)
            .first()
            .timeout(15000)
            .subscribe((authResponse) => {
                console.log("next fired");
                dialogs.alert("Next: " + authResponse);
            }, (error) => {
                //fired only after 15s when mobile is empty
                console.log("error fired");
                console.log(JSON.stringify(error));
            }, () => {
                console.log("completed fired");
            });

注册()

public signUp(user: User): Observable<AuthResponse> {
        let observable = new Subject<AuthResponse>();
        this.isUserExisting(user.mobile)
            .first()
            .subscribe((isUserExisting) => {
                if (isUserExisting) {
                    console.log("User already exists");
                    observable.next(AuthResponse.USER_EXISTING);
                } else {
                    console.log("User does not exist");
                    this.saveUser(user).first().subscribe(() => {
                        observable.next(AuthResponse.SUCCESS);
                    })
                }
            }, (error) => {
                return Observable.throw(error);
            })
        return observable;
    }

    public isUserExisting(mobile: string): Observable<boolean> {
        let observable = new Subject<boolean>();
        if (!mobile) {
            console.log("empty mobile");
            return Observable.throw(new Error("Mobile number cannot be empty"));
        }
        firebase.query(() => { }, "/users",
            {
                singleEvent: true,
                orderBy: {
                    type: firebase.QueryOrderByType.CHILD,
                    value: "mobile"
                },
                range: {
                    type: firebase.QueryRangeType.EQUAL_TO,
                    value: mobile
                }
            }
        ).then((result) => {
            console.log("Checking for user success: ");
            console.log(JSON.stringify(result));
            observable.next(result.value != null);
        });
        return observable;
    }

更新: 保存用户()

public saveUser(user: User) {
        return Observable.defer(() => firebase.push('/users', user)
            .then((result) => {
                console.log("Created user record with key: " + result.key);
                console.log("Dumping result:");
                console.log(JSON.stringify(result));
            })
            .catch((error) => {
                console.log("Error while saving user: " + error);
            })
        )
    }

Answer 1

创建自己的主题是一种不好的做法，您将在该主题上发出由内部订阅的可观察值发出的可观察值。这会导致内存泄漏，因为您无法取消内部订阅。

public signUp(user: User): Observable<AuthResponse> {
  return this.isUserExisting(user.mobile)
    .flatMap(isExistingUser => {
      if (isExistingUser) {
        console.log('user already exists: ' + isUserExisting);
        return Rx.Obserable.of(AuthResponse.USER_EXISTING);
      }
      return saveUser(user).map(result => AuthResponse.SUCCESS)
    })
    .catch(err => {
      console.log(`error during signup of user: ${user.mobile})`);
      return Rx.Observable.of(AuthResponse.FAILED_TO_SIGNUP);
    })
    .first();
}

如果没有 signUp 函数中的订阅，您的代码可以取消订阅，并且错误会在您这边不做任何工作的情况下传播。这也是您超时的原因而不是检查您的手机号码；您忘记传播错误。相反，您尝试从 subscribeOnError 回调中返回 Rx.Observable.throw，该回调的签名为 onError(error):void

一个好的做法是让你的 Rx 函数保持惰性；仅当有人订阅时才执行其中的代码。这有助于减少难以追踪的错误。在您使用 firebase.query() (返回 promise )的情况下，您可以使用 .defer() 等待执行 promise ，直到有人订阅您的 isUserExisting 函数。

public isUserExisting(mobile: string): Observable<boolean> {
  if (!mobile) {
      console.log("empty mobile");
      return Observable.throw(new Error("Mobile number cannot be empty"));
  }

  return Rx.Observable.defer(() => firebase.query(() => { }, "/users",
      {
          singleEvent: true,
          orderBy: {
              type: firebase.QueryOrderByType.CHILD,
              value: "mobile"
          },
          range: {
              type: firebase.QueryRangeType.EQUAL_TO,
              value: mobile
          }
      }
  ))
  .do(firebaseResult => console.log('Checking for user success: ' + JSON.stringify(firebaseResult)))
  .filter(firebaseResult => firebaseResult.value != null);
}

public saveUser(user: User) {
  return Observable.defer(() => firebase.push('/users', user))
    .do(result => console.log(`created user with key ${result.key}`))
}