RxJS: How to delay next emitted value?-CodePudding

I have async function handleParamsChanges which can take a few seconds to resolve. And I'm calling it when observable emits value:

this._activatedRoute.params
  .subscribe(params => {
    this.handleParamsChanges(params).then(() => {
      // new value can be processed now
    });
  });

How could I modify my code so that if observable emits 2 values one after another, first handleParamsChanges is called for first value, and only after this promise resolves, it's called with second value, and so on.

Edit:

Here is the solution I came up with, but I'm guessing there is a better way to do this:

const params$ = this._activatedRoute.params;
const canExecute$ = new BehaviorSubject(true);

combineLatest(params$, canExecute$)
  .pipe(
    filter(([_, canExecute]) => canExecute),
    map(([params]) => params),
    distinctUntilChanged()
  )
  .subscribe(async params => {
    canExecute$.next(false);
    try {
      await this.handleParamsChanges(params);
    } catch (e) {
      console.log(e);
    } finally {
      canExecute$.next(true);
    }
  })

I'm using canExecute$ to delay processing of new value.

I need to use distinctUntilChanged here to avoid creating infinite loop.

CodePudding user response：

What you're looking for is concatMap. It waits for the previous "inner observable" to complete before subscribing again. Also you can greatly simplify your pipe:

params$.pipe(
  concatMap(params => this.handleParamsChanges(params)),
).subscribe()