Do operation on two Observables in angular

I have a service which gets two observables, storeValueOne$ and storeValueTwo$, which are of type number and update frequently but not at the same time.

I want to add the values of storeValueOne, storeValueTwo and store them into another observable which keeps track of the changes.

With ngrx merge I can combine the two observables together but it's not clear to me how to do the calulcation as they are updating at different rates, and I also don't know which one changed its value.

import { Injectable } from '@angular/core';
import { forkJoin, map, merge } from 'rxjs';
import { UnstableCompComponentTwoStore } from './components/unstable-comp-two/unstable-comp-two.store';
import { UnstableCompStore } from './components/unstable-comp/unstable-comp.store';

@Injectable({
  providedIn: 'root'
})
export class GlobalStateService {

  constructor(
    private storeUnstableOne: UnstableCompStore,
    private storeUnstableTwo: UnstableCompComponentTwoStore
  ) { }

  storeValueOne$ = this.storeUnstableOne.select((state) => state.unstableStateVal);
  
  storeValueTwo$ = this.storeUnstableTwo.select((state) => state.unstableStateValTwo);


      combineData(){
        const merged = merge(this.storeValueOne$, this.storeValueTwo$)

        // merged.forEach(d => console.log(d))
        /* value 1 + value 2 */

        // mergedState = value1 + value2 

      }

}

Solution 1:

Sounds like you need combineLatest:

import { combineLatest } from 'rxjs'; 

const combined = combineLatest([obs1, obs2]);

combined
  .subscribe(([latest1, latest2]) => console.log({ latest1, latest2 }));

Note: the combined observable will only emit once every member has emitted at least once.
From that point on, it will emit every time a member emits, giving you latest values from each observable, maintaining the original order of the combined observables (so first value in the array will be the last value of the first combined observable).

Visualize it here.

In the example above, if you want to turn it into an observable of the total of latest values from each contained observable:

combined.pipe(
  map(values => values.reduce((o, k) => +o+k, 0))
)