角2个可观测量 - 需要等待，直到操作完成运行前可观察

Question

我一直在慢慢变观测的窍门，虽然火力地堡似乎需要你为了得到所取得的成果做了很多疯狂的东西与他们： 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的所有逻辑请试试看，并记住你必须退订observables！

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(); 

     }); 
    }); 
    }