How can I give an RxJS observable pipe access to the original observable's emission AND the pipe's previous emission?

Question

I have an RxJS Observable that emits a series of changes to an underlying data structure—specifically, snapshotChanges() from an AngularFirestoreCollection.

I'm currently mapping this to an array of plain JavaScript objects for consumption by my app.
This array is not protected in any way, and consuming code could accidentally modify this structure.
The entire array is rebuilt whenever the underlying data source emits, even if only one (or sometimes no) item in the array has actually changed.
Because of this, all references change each time, making change detection harder than it needs to be—and really slowing down my app.

What I want to do instead is use Immer to maintain an immutable structure, such that unchanged data is structurally shared with the “new” array.

What I can't work out is how to pipe() off the snapshotChanges() observable such that the pipe gets access to the previously emitted immutable data (or a first-time default) in addition to the latest snapshotChanges() output.

In code, what I basically already have is this:

const docToObject = (doc) => { /* change document to fresh plain object every time */ };
const mappedData$ = snapshotChanges().pipe(
    map(changes => changes.map(change => docToObject(change.payload.doc)),
    tap(array => console.log('mutable array:', array)),
);

and I'm essentially looking for something like this, where I don't know what XXX(...) should be:

const newImmutableObject = (changes, old) => {
  // new immutable structure from old one + changes, structurally sharing as much as
  // possible
};
const mappedData$ = snapshotChanges().pipe(

// ==================================================================================
    XXX(...), // missing ingredient to combine snapshotChanges and previously emitted
              // value, or default to []
// ==================================================================================

    map(([snapshotChanges, prevImmutableOutput]) => newImmutableOutput(...)),
    tap(array => console.log('IMMUTABLE ARRAY with shared structure:', array)),
);

I feel like the expand operator is close to what I need, but it seems to only pass the previously emitted value in on subsequent runs, whereas I also need the newly emitted snapshotChanges.

Given an RxJS Observable pipe, how can I operate on this Observable's emissions while also having access to the pipe's previous emission?

scan seems to be what you are looking for: snapshotChanges().pipe(scan((prev, changes) => newImmutableObject(changes, prev), [])) — cartant
Thank you—this is exactly what I was looking for (I was so close!). The learn-rxjs page for scan even says “You can create Redux-like state management with scan!” Example 2: Accumulating an object shows almost precisely what I'm trying to achieve. — Alex Peters

Sunil Singh Sunil Singh · Accepted Answer · 2018-11-10T09:41:04

As per your requirement I would suggest to use scan operator which can track all previous state and new state.

const newImmutableObject = (changes, old) => {
  // new immutable structure from old one + changes, structurally sharing as much as
  // possible
};
 const mappedData$ = snapshotChanges().pipe(
 scan((acc, current) => [...acc, current], []), //<-- scan is used here
 map(([snapshotChanges, prevImmutableOutput]) => newImmutableOutput(...)),
    tap(array => console.log('IMMUTABLE ARRAY with shared structure:', array)),
);

How can I give an RxJS observable pipe access to the original observable's emission AND the pipe's previous emission?

1 Answers