Join two observable arrays together

Question

checkAllThreads() {
    const userId = this.auth.getUser().uid;
    const buyerThreads = this.afs.collection('threads', ref => ref.where('buyerId', '==', userId)).valueChanges();
    const sellerThreads = this.afs.collection('threads', ref => ref.where('sellerId', '==', userId)).valueChanges();
    this.subscription = forkJoin(buyerThreads, sellerThreads)
      .map(([bT, sT]) => [...bT, ...sT])
      .subscribe(res=> { 
        console.log(res) //never console logs
        //logic to iterate through array, look for unread threads goes here
      });
  }

I'm using AngularFire2 to query my Firestore database for all documents in the 'threads' collection with the fields sellerId or buyerId equal to their userId. From what I understand, AngularFire2 does not allow you to query all documents where either one is true; if you chain their .where() method, it will return all documents where both are true (which in my case would be 0). So, I'm trying to get two Observable arrays and combine them in one single array in order to tell the user in real time whether or not they have unread messages.

I tried the code above (adapted from an answer here), which didn't run anything in the subscribe method. From what I can tell, I think it's because my observables aren't completed, as I need to have the data update in real time. Is there any way to take two still-streaming observable arrays and combine them into one, or will I have to figure out another way?

Answer 1

Use combineLatest instead of forkJoin.

So it should be...

checkAllThreads() {
    const userId = this.auth.getUser().uid;
    const buyerThreads = this.afs.collection('threads', ref => ref.where('buyerId', '==', userId)).valueChanges();
    const sellerThreads = this.afs.collection('threads', ref => ref.where('sellerId', '==', userId)).valueChanges();
    this.subscription = Observable.combineLatest(buyerThreads, sellerThreads)
      .map(([bT, sT]) => [...bT, ...sT])
      .subscribe(res=> { 
        console.log(res);
      });
  }

You may need to import these two lines.

import 'rxjs/add/observable/combineLatest';
import { Observable } from 'rxjs/Observable';