How does RxFire handels firestore queries

Question

This js reduce works fine to handle the query result:

function toc(current) {
  return {....};
};
function getToc(data) {
  return = data.reduce((a, c) => Object.assign(a, {[c.id]: toc(c)}), {});
};

const query = db.collection(normCollection)
.where('a_id', '==', a_id )
.where('year', '==', year) 
.orderBy("id");
subscriptionNorm = collectionData(query, "id")
.subscribe(data => console.log(getToc(data)));

But when I use RxJs reduce, it stops working. It has something to do with the stream end, but ... But I do not understand how RxFire / RxJs handles a streamed firestore query result:

...
subscriptionNorm = collectionData(query, "id")
.pipe(reduce((a, c) => Object.assign(a, {[c.id]: toc(c)}), {}))
.subscribe(data => console.log(data));

Update this works fine, but ...:

...
subscriptionNorm = collectionData(query, "id")
.pipe(
  map(v => v.reduce((a, c) => 
    Object.assign(a, {[c.id]: toc(c)}), {})
  ),
)
.subscribe(data => console.log(data));

Answer 1

Your assumption is correct about the rxjs reduce operator.

"Applies an accumulator function over the source Observable, and returns the accumulated result when the source completes" - from the docs, see here: RxJS reduce docs

In your case source won't complete because that's how Firestore works, it is running endlessly without completion, until error happens or you unsubscribe manually.

For a rough example, you could use take(1) operator inside the pipe and it will complete the source Observable after emitting 1 event, thus the reduce will work, but it kills the main idea behind the Firestore Observable.

This is the way you can utilize rxjs reduce operator:

subscriptionNorm = collectionData(query, "id").pipe(
  switchMap(data => from(data).pipe(
    reduce((a, c) => Object.assign(a, { [c.id]: toc(c) }), {})
  )),
).subscribe(data => console.log(data));

This is possible because I am switching to the from(data) and the inner Observable will get completed, thus the reduce operator will work as you intend.

But, to be honest that's an overkill and you could simply keep the way you already have implemented:

subscriptionNorm = collectionData(query, "id").pipe(
  map(data => data.reduce((a, c) => Object.assign(a, { [c.id]: toc(c) }), {}))),
).subscribe(data => console.log(data));