合并多个 observables 和 return 一个

Question

我有一个函数应该 return 一个带有 Conversations 列表的可观察对象。为此，我必须从另一个 table Participants 获取用户对话 ID，并将这些获取的内容一个一个地组合起来。

public getConversations(user_id: string): Observable<Conversation[]> {
    return this.firebase
      .collection<Participant>(this.PARTICIPANTS_COLLECTION, ref =>
        ref.where('user_id', '==', user_id)
      )
      .valueChanges()
      .map((participants: Participant[]) => {
        const conversation_ids: string[] = [];
        participants.forEach((p: Participant) => conversation_ids.push(p.conversation_id));
        return conversation_ids;
      })
      .mergeMap((ids: string[]) => {
        let conversations: Observable<Conversation[]> = new Observable<Conversation[]>();
        for(let id of ids) {
          conversations.merge(
            this.firebase.collection<Conversation>(this.CONVERSATIONS_COLLECTION, ref =>
              ref.where('id', '==', id)
            ).valueChanges() // I want to merge these
          );
        }
        return conversations; // I want to return this
      });
  }

我应该使用哪个运算符来组合这些可观察值？

Answer 1

merge() 不适用于 firebase，因为 Observable .valueChanges() 永远不会完成，并且 merge() 需要它合并的所有 observable 完成才能发出结果值。来自 docs:

merge subscribes to each given input Observable (as arguments), and simply forwards (without doing any transformation) all the values from all the input Observables to the output Observable. The output Observable only completes once all input Observables have completed. Any error delivered by an input Observable will be immediately emitted on the output Observable.

您将需要使用 Observable.combineLatest()。此外，您可以使用 Javascript 的原生 map() 来消除代码中的 forloop：

return this.firebase.collection<Participant>(this.PARTICIPANTS_COLLECTION, ref => ref.where('user_id', '==', user_id)).valueChanges()
    .map((participants: Participant[]) => participants.map(p => p.conversation_id))
    .mergeMap((ids: string[]) => Observable.combineLatest(ids.map(id => this.firebase.collection<Conversation>(this.CONVERSATIONS_COLLECTION, ref => ref.where('id', '==', id)).valueChanges())));

合并多个 observables 和 return 一个

Merge multiple observables and return one

observable

rxjs

firebase

angularfire2

angular