RxJs: Prevent infinite loop

Question

Consider the following scenario:

observerX => observableA <= observerY => observableB <= observerZ

There are two Observables (A and B) and three Observers (X, Y, and Z).
Each one of those Observers and Observables are within separate (Angular) services.
Observer Y observes both Observable A and Observable B, which stay "alive" over the life of the application, i.e. they do not complete.
When Observable A triggers an event, Observer Y calls a function on the service of Observable B, which causes Observable B to trigger an event. Similarly, when Observable B triggers an event, Observer Y calls a function on the service of Observable A, which in turn causes Observable A to trigger an event.

Ex: Observer Y...

svcA.obsA.subscribe(()=> {
    svcB.doSomething()
});

svcB.obsB.subscribe(()=> {
    svcA.doSomething()
});

As you can see, all Observer Y is trying to do is to keep service A and service B in sync.

The problem: The setup as described above causes an infinite execution loop. One idea that came to my mind to get around this issue is to set a flag in each subscribe, and when that flag is set the other subscribe will ignore the next event. That is not a proper solution however because the next event may not necessarily be the one that corresponds to Observer Y calling doSomething(), but may be due to Observer X or Z calling doSomething().

Is there an elegant solution to this use-case? Do I have an architectural issue here? Please let me know if my description of the problem is not clear.

JSFiddles:

Problem example
Monkey patched example (only ObserverY class has changed in this example)

Can you specify problem more concretely? I.e. what are you doing the the data from Observable A and B? What kind of data are we taking about, and what is the relationship forcing you to sync two Observables? Maybe you could use jorkJoin? — Siegmeyer
@Siegmeyer thank you for your response. I will need to get back to you with more details in a couple of days when I'm back at work. — Sam Herrmann
@Siegmeyer I have added JSFiddle examples that will hopefully help clarify the problem. You will see that the observables do not return any data, they simply notify the observers that a task was completed. The reason for this setup between service A and service B is to keep them modular and completely decoupled from each other. It is the responsibility of ObserverY (application logic) to make sure that when one of the services is poked to perform their task that the other service does its part as well. I hope this helps. — Sam Herrmann

Mojtaba Mojtaba · Accepted Answer · 2020-12-11T14:35:49

I think you can use thottletime operator (https://www.learnrxjs.io/learn-rxjs/operators/filtering/throttletime) with a short duration selector e.g, 100ms for both observable A and B.

obsA.pipe(throttleTime(100)).subscribe(val => obsB.next(val));
obsB.pipe(throttleTime(100)).subscribe(val => obsA.next(val));

Asumming that the second emits of observables A and B are less than 100ms,they will be ignored.

RxJs: Prevent infinite loop

2 Answers