How to combine multiple Observable type arrays Observable<T[]>?

0
votes

I'm running multiple queries on Firebase Firestore. They all (can) return an Observable. I would like to combine all Observable into ONE Observable.

I have tried different RxJS operators, like; combineLatest, forkJoin, concat, zip, merge and mergeMap. But I have not been able to achieve the desired result.

getPrivateBooksFromAuthors(authors): Observable<Book[]> {
    let bookRefs: Observable<Book[]>[] = [];
    authors.map(key => {
      bookRefs.push(this.afs.collection<Book>('books', ref => 
        ref.where('public', '==', false)
           .where('authors', 'array-contains', key)
        ).valueChanges())
    });
    return bookRefs[0]
}

In the above piece of code I get all the private books from authors[0]. When I return concat(...bookRefs) or concat(bookRefs[0], bookRefs[1]), I still only get the books from authors[0]. I expect to get all the books from all the authors provided.

3
angulartypescriptrxjsgoogle-cloud-firestore

3 Answers

2
votes

As @Doflamingo said, you can use forkJoin in order to call them in parallel and receive a response with an array of all responses resolved. The problem you have is that every response is a Book[], so in the forkJoin you receive an array of Book[] (Book[][]). You need to flat the Book[][] into a Book[]. You can do it using the a map an a reduce function.

Here I posted a simple code snipped that console logs the response of the forkJoin and the result after applying the reduce function.

function mockBooks$(key): Observable<Book[]> {
  return rxjs.of([key + '-book1', key + '-book2', key + '-book3']);
}

function getPrivateBooksFromAuthors(authors): Observable<Book[]> {
    let bookRefs: Observable<Book[]>[] = authors.map(key => mockBooks$(key));

    return rxjs.forkJoin(bookRefs).pipe(
        rxjs.operators.tap((books) => console.log('After forkJoin', books)),
        // You need this flattening operation!!
        rxjs.operators.map(books => books.reduce((acc, cur) => [...acc, ...cur], []) ));
}


const authors = ['a1', 'a2', 'a3', 'a4', 'a5'];
getPrivateBooksFromAuthors(authors).subscribe((data) => console.log('Final result', data));
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.js"></script>

Hope this helps!

1
votes

I think the best solution is use fork join.

Fork let you to execute the call in parallel. The result of every call it's put in one object (join) and you can get the all call's result.

1
votes

@Doflamingo pointed us in the direction of forkJoin and @Llorenç gave a great code example with the map and reduce pipes to combine the Observables into one Observable. For me, the forkJoin failed if some authors didn't have Books. So I ended up going with combineLatest (which doesn't wait for all the Observables to be completed). This is my final code (all the credit goes to @Llorenç):

  mockBooks$(key): Observable<Book[]> {
    return this.afs.collection<Book>('books', ref => 
      ref.where('private', '==', true)
         .where('authors', 'array-contains', key)
      ).valueChanges()
    // return of([key + '-book1', key + '-book2', key + '-book3']);
  }

  getPrivateBooksFromAuthors(authors): Observable<Book[]> {
    let bookRefs: Observable<Book[]>[] = authors.map(key => this.mockBooks$(key));

     // return combineLatest(bookRefs).pipe(
     //     tap((books) => console.log('After forkJoin', books)),
     //     // You need this flattening operation!!
     //     map(books => books.reduce((acc, cur) => [...acc, ...cur], []) ));

     return combineLatest<Book[]>(bookRefs).pipe(
        map(arr => arr.reduce((acc, cur) => acc.concat(cur) ) ),
     )
  }

I've commented out Llorenç's code, to show the difference. Thank you to both of you!