How to compute a new result whenever any of the given source streams changes?

[jzarzeckis]

If I need an observable that computes result depending on other result - I will very often use combineLatest(), but when I've reached a point where I already have multiple streams that are created with combineLatest() they will be emitting too often, and the intermediate results may even be what I don't want in the output.

const a = new Subject();
const b = new Subject();

// Should compute a + b from latest a and b values
const abSum = combineLatest([a, b]).pipe(map(([a, b]) => a + b));

// Should compute `a + 2 * b` when a or b change
const abbSum = combineLatest([abSum, b]).pipe(map(([a, b]) => a + b));

abbSum.subscribe((x) => {
  console.log(x);
});

a.next(5);
b.next(7);
// Logs 19 (which is expected `5 + 7 * 2 = 19`)

b.next(3);
// Logs 15 - and then 11. But 15 is basically an invalid state to me

Perhaps anyone has encountered a similar issue, and has an idea how to avoid the invalid intermediate state, when using too many combineLatest together?

In my case - the final sum should be the same as a + 2 * b. And when I change b from 7 to 3, I only want to get the output which is 5 + 3 * 2 = 11. But because of my combineLatest operators - 2 different events are created, where the initial one produces an incorrect result. Is there some guideline to follow to avoid such discrepancies? (So far I only imagined that auditTime(0) could always ensure that the wrong intermediate states are skipped. But introducing timers to solve a problem in logic feels very dirty to me)

[jzarzeckis]

One solution I can imagine is to rewrite it as:

const abbSum = abSum.pipe(withLatestFrom(b), map(([a, b]) => a + b));

But this requires me to know that abSum will emit always together with b. And if abSum had some filter applied or more complex logic - I could still encounter invalid state, where sources changed, but final result didn't get updated..

[voliva]

More importantly, you would need withLatestFrom(b) to receive the updated value from b before abSum receives it.

This won't work in the following scenario:

const a = new Subject();
const b = new Subject();

// Should compute a + b from latest a and b values
const abSum = combineLatest([a, b]).pipe(
  map(([a, b]) => a + b),
  share() // Share the subscription
);

// Someone else subscribes. `b` will first emit to the combineLatest above
abSum.subscribe();

// Should compute `a + 2 * b` when a or b change
const abbSum = abSum.pipe(
  withLatestFrom(b), // This subscription to `b` comes in after the subscription from the combineLatest above
  map(([a, b]) => a + b)
);

abbSum.subscribe((x) => {
  console.log(x);
});

a.next(5);
b.next(7);
// Logs nothing!
// `b` has emited to the combineLatest, which causes `abSum` to emit [12]
// `abbSum` picks that value up, but its withLatestFrom hasn't received a value from `b` yet, so it drops it.
// `b` then emits to abbSum's withLatestFrom

b.next(3);
// Logs 15 (incorrect value)
// `b` has emited to the combineLatest, which causes `abSum` to emit [8]
// `abbSum` picks that value up, but its withLatestFrom has the previous value from b=7.
// `abbSum` emits 8+7 = [15]
// `b` then emits to abbSum's withLatestFrom

I don't have a perfect solution, and it's something I've struggled it with myself. I think withLatestFrom is the cleanest one, but you must be careful not to share subscriptions to other places or you might get these kind of glitches. Or I also use debounceTime(0) (which I think it's the same as auditTime for this case)

I'd be really surprised if a solution is found. The problem is that you have 2 observables you want to combine, but those observables might emit at the same event loop (synchronously) or they might not. When you receive an emission from one of them, what do you do?

• If you don't wait, then maybe the second observable might emit synchronously so you'll have to compute twice.
• If you try to wait (by setTimeout or a microtask), then maybe the second observable doesn't emit synchronously and you have added asynchrony on your computation.

[jzarzeckis]

Well. share() is an entirely other can of worms - it can affect behaviour in a lot of other scenarios too. Was hoping to keep that out of the scope of the problem here :)

I guess I'll play with auditTime() + asapScheduler

[jzarzeckis]

Actually it's not that bad. And even passes in the marble tests:

const a = cold('a', { a: 5 });
const b = cold('a-b', { a: 7, b: 3 });

const abSum = combineLatest([a, b]).pipe(map(([a, b]) => a + b));
const abbSum = combineLatest([abSum, b]).pipe(
    auditTime(0, asapScheduler),
    map(([a, b]) => a + b)
);

expectObservable(abbSum).toBe('a-b', { a: 19, b: 11 }); // PASS

[jzarzeckis]

Just so this can be closed as "answered", let me mark the IMHO correct approach in a separate comment:

const a = cold('a', { a: 5 });
const b = cold('a-b', { a: 7, b: 3 });

const abSum = combineLatest([a, b]).pipe(map(([a, b]) => a + b));
const abbSum = combineLatest([abSum, b]).pipe(
    auditTime(0, asapScheduler),
    map(([a, b]) => a + b)
);

expectObservable(abbSum).toBe('a-b', { a: 19, b: 11 }); // PASS