CODEWITHSUNDEEP
angulartypescriptrxjsobservablebehaviorsubject
Maurice
Maurice

Reputation: 7381

How do i prevent behaviorsubject next() calls from overwriting each other?

I have a BehaviorSubject variable called saveToLocalStorage and in one method the next() method is called twice. This is causing a problem because only one of the two is ever completed. The other one is simply overwritten. The service that is subscribed to the calls saves the data to localstorage nothing else. Apparently there has to be a delay between both next calls so i decided to handle the calls inside an Observable like so (lets assume that the if statement returns true):

Observable.of(1).pipe(
  tap(()=> {if(this.currentWebAppMode !== newMode){
    this.saveToLocalStorage.next([DataType.STORED_RESULTS, null])}; 
  }),
  debounceTime(100))
  .subscribe(result => this.saveToLocalStorage.next([DataType.WEB_APP_MODE, newMode]));

This works, but feels really hacky. So i was wondering whether there is a more proper way to make sure both subsequent next() calls to saveToLocalStorage are able to execute properly.

Edit: Thanks to Deborah i was able to ameliorate my solution by incorporating the concatMap operator instead of tab and debounceTime Heres my new code:

Observable.of(1).pipe(
  concatMap(()=> {if(this.dataStoreService.currentAppMode !== newMode){
    this.saveToLocalStorage.next([DataType.STORED_RESULTS, null])};
  return of(true)}))
  .subscribe(result => this.saveToLocalStorage.next([DataType.WEB_APP_MODE, newMode]));

Upvotes: 1

Views: 1945

Answers (1)

DeborahK
DeborahK

Reputation: 60518

Using a concatMap will ensure that each source value is processed in a serial fashion.

The code below first creates an "action" stream that emits the values (I'm just emitting 1 and 2 but you get the idea).

As each item is emitted, the code executes the desired pipeline.

Within the pipeline, the code uses concatMap to ensure each emitted item is fully processed. If an item is emitted before the first item's processing is done, the second item is cached internally and will be processed as soon as the first emitted item's processing is complete.

In this example, the code calls doSave to actually perform the saving to local storage process ... which emits true if the save was successful.

Something like this:

  saveSubject = new BehaviorSubject<number>(1);
  saveAction$ = this.saveSubject.asObservable();

  source = this.saveAction$.pipe(
    concatMap(value =>
      this.doSave(value)
    )
  );

  ngOnInit() {
    this.source.subscribe();
    this.saveSubject.next(2);
  }

  doSave(value) {
    console.log('Emited value:', value);
    // Save to local storage
    return of(true);
  }

I have a stackblitz here: https://stackblitz.com/edit/angular-concatmap-deborahk

Upvotes: 1

Related Questions