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

     }); 
    }); 
    }