How to prevent concurrent effect execution

Question

I have an expensive calculation that is called by an effect. I now want to ensure, that this calculation is never called concurrently, i.e. if it is called a second time while the first call is still running, the second call should be ignored.

My approach to this problem was to create 2 actions: calculate and setLoading.

@Effect()
calculate$ = this.updates$
  .whenAction(CALCULATE)
  .flatMap(data => {
    console.debug('LOADING', data.state.loading);
    if (!data.state.loading) {
      this.store.dispatch(Actions.setLoading(true));
      await DO_THE_EXPENSIVE_CALCULATION();
      this.store.dispatch(Actions.setLoading(false));
    }
  });

with Actions.setLoading obviously setting state.loading. However, if I start the calculation 2 times in a row:

store.dispatch(Actions.calculate());
store.dispatch(Actions.calculate());

the output is

LOADING false
LOADING false

and therefore, the expensive calculation is executed two times. How can I prevent this?

Answer 1

You might see the LOADING false twice because the Action.setLoading have not been executed yet. This is very possible depending on the synchrony/asynchrony of dispatching and actions. Best is to not make hypothesis about that.

Generally speaking, if you want to limit a number of operations executing at the same time/execute only one operation at a time, there are a number of operators you could use in rxjs v4/v5:

1. flatMapWithMaxConcurrent|mergeMap : will subscribe to the parameterized maximum of observables concurrently, keep a buffer of the remaining observables to subscribe to, and subscribe to them when a slot becomes available. Hence there is no loss.

2. flatMapFirst|exhaustMap : will only subscribe to one observable at a given time. Observable which come while the current observable is being executed are lost. When the current observable is completed, new observables can be subscribed to.

3. concatMap : will subscribe only to one observable at a time. Will keep a buffer of remaining observables to subscribe to, and will do that in order when the current observable has completed. Hence there is no loss.

You can also review the following question : Batching using RxJS?

In conclusion, maybe something like this could work for you :

@Effect()
calculate$ = this.updates$
  .whenAction(CALCULATE)
  .exhaustMap(data => DO_THE_EXPENSIVE_CALCULATION())
  ;

I supposed here that the DO_THE_EXPENSIVE_CALCULATION() returns a promise (could also be an observable).