Combine initial Observable with an Observable for changes (into a single Observable)

Question

I have two observables. One that fetches the initial data, and another one that should react to changes and applay them accordingly:

const initial$: Observable<Report[]>;
const changes$: Observable<ObjectChangeEvent<Report>>;

Some characteristics are:

1. initial$ has to complete first, before changes can be applied
2. changes$ can emit 0..n times
3. the changed array should be the basis for the next change emit. That means only the first change should apply to the initial State. Following changes should not discard previous ones.

I want to combine both observables into a single one. The closest I've come so far, is the combineLatest Operator. But it conflicts with characteristic 2), as changes$ may not emit anything.

Any help will be greatly appreciated.

Answer 1

Do you mean somehting like this:

this.result$ = this.changes$.pipe(
  skipUntil(this.initial$),
  // ...
);

Or...

const initial$: Observable<Report[]>;
const changes$: Observable<ObjectChangeEvent<Report>>;

const accumulatedChanges$: Observable<ObjectChangeEvent<Report>[]>;

this.accumulatedChanges$ = this.changes$.pipe(
  scan((acc, curr) => [...acc, curr], []),
  startWith([]),
); // emits [change1], [change1, change2], [change1, change2, change3]....

this.result$ = combineLatest([this.initial$, this.accumulatedChanges$]).pipe(
  // apply all accumulated changes on initial
);

---

Edit:

OR....

this.result$ = this.initial$.pipe(
  switchMap(initial => this.changes$.pipe(
    // applyChangeToCurrent is only a placeholder for your semantic to alter the current
    scan((current, change) => applyChangeToCurrent(current, change), initial),
  )),
);

Answer 2

To satisfy the first requirement you can use last operator, combining with mergeMap to switch to the changes$ observable and using the last value of initial$ in it. Using startWith operator using it as initial state in case changes$ is empty.

Generally it should look something like this:

initial$.pipe(
  last(),
  mergeMap(initialState =>
    changes$.pipe(startWith(initialState))
  )
);