Angular 2 Observables - 在运行 Observable 之前需要等到操作完成

Question

我已经慢慢掌握了 observables 的窍门，尽管 Firebase 似乎需要你对它们做很多疯狂的事情才能获得结果：P 基本上，我有一个函数可以过滤掉一些键，基于某些条件，以便创建一组用户以从我的 Firebase 数据库中获取。

但是，Observable 是在我的过滤功能完成之前运行的，所以我遇到了竞态条件。我的函数有点复杂（至少对我来说），所以我不确定我能做些什么来确保 userKeys 在Observable.combineLatest 调用之前是最新的：

getUsersForConversations(conversations) {
    conversations.forEach((conversation, index) => {

      // Get user out of local storage
      this.storage.get('user').then(user => {

        // Iterate through the users and remove the current user
        // to prevent an unecessary call being made
        let userKeys = Object.keys(conversation.users);

        userKeys.forEach((key, index) => {
          if(key === user.id) {
            userKeys.splice(index, 1);
          } else {

            if(userKeys.length > 0) {

              // Grab the conversation for this user and determine
              // If the conversation has been removed before
              this._af.database
              .object(`/social/conversations_last_deleted/${conversation.$key}/${user.id}`)
              .subscribe(lastDeleted => {

                if(lastDeleted && (lastDeleted.$value !== null)) {
                  this._messages.getMessagesForConvo(conversation.$key, lastDeleted).subscribe(messages => {

                    if(messages.length === 0) {
                      userKeys.splice(index, 1); // This is updated after the Observable.combineLatest :(
                    }
                  })
                };
              });
            }
          }
        });

        // Get all the users based on this list and shove them
        // into the correct conversation

        Observable.combineLatest(
          userKeys.map((userKey) => this._af.database
            .object(`/social/users/${userKey}`)
          )
        ).subscribe(users => {
          conversations[index].users = users;
          this.conversations = conversations;
        })
      });
    });
  }

有人有什么想法吗？

Answer 1

好吧，我尝试在 Observable 中执行所有逻辑，请试一试并记住您必须取消订阅 observable！

  getUsersForConversations(conversations) {
    conversations.forEach((conversation, index) => {

      // Get user out of local storage
      this.storage.get('user').then(user => {

        let userKeys = Object.keys(conversation.users);

        let sub = Observable
          .from(userKeys)
          .filter(key => (key != user.id && key.length > 0))
          .flatMap(key =>
            Observable.zip(
              Observable.of(key),
              this._af.database.object(`/social/conversations_last_deleted/${conversation.$key}/${user.id}`)
                .filter(lastDeleted => (lastDeleted && (lastDeleted.$value !== null)))
                .flatMap(lastDeleted => this._messages.getMessagesForConvo(conversation.$key, lastDeleted)),
              (key, messages) => {
                if (messages.length === 0) {
                  return null;
                }
                return key;
              })
          )
          .filter(key => key != null)
          .flatMap(userKey => this._af.database.object(`/social/users/${userKey}`))
          .subscribe(users => {
            conversations[index].users = users;
            this.conversations = conversations;
          });

          // REMEMBER TO ALWAYS UNSUBSCRIBE!!!! or you will have memory leaks....
          sub.unsubscribe();

      });
    });
  }