How can I update an Observable without losing active subscribers?

user_12221382 0
Viewed 824

I'm currently experiencing a bit of trouble trying to update a cached value using publishReplay and refCount.

First off, the method in question

getCurrentUser(): Observable<User> {
    if (!this.jwtService.getToken() || this.jwtService.getAnonymousToken()) { return of(null); }
    if (!this.currentUser$) {
      console.warn('user.service.getcurrentuser -> load and CACHE it once');
      this.currentUser$ = this.userService.loadCurrentUser().pipe(
        // https://medium.com/angular-in-depth/fastest-way-to-cache-for-lazy-developers-angular-with-rxjs-444a198ed6a6
        publishReplay(1),
        refCount()
      );
    }
    console.log('user.service.getcurrentuser -> get from CACHE');
    return this.currentUser$;
 }

this.currentUser$ is an Observable of type User.

The method itself works as intended, the first call of the method loads the current user from the server. Subsequent calls then get the cached entry. What I'm trying to do now, however, is to remove the cached value and reload the user from the server (ie. something was edited on the frontend and the user data requires reloading).

That I achieved by doing

removeCachedUser(): void {
  this.currentUser$ = null;
}

By doing this the if condition in the getCurrentUser method is triggered and the user is reloaded from the server. So far, so good.

The problem I'm facing is that methods that subscribed to getCurrentUser before the reset of the cache aren't receiving any updates, ie

ngOnInit() {
  this.currentUser$ = this.authService.getCurrentUser();
}

And my suspicion is that it's because of my overwriting the this.currentUser$ Observable.

this.currentUser$ = this.userService.loadCurrentUser()

This post more or less confirms my theory I suppose.

Am I right? And if so, what's the proper way to handle this issue? Unfortunately, I don't have a reactive backend so I have to trigger the reloads manually. Thanks in advance!

rxjs angular typescript
3 Answers

You have to create an Observable that can emit multiple times to be able to send new values to active subscribers. In your current setup this.currentUser$ completes when this.userService.loadCurrentUser() completes. I suspect loadCurrentUser is a http request that emits once and completes. So this.currentUser$ can't emit multiple times.

To create an Observable that can emit multiple times use a Subject as the source and map to your http request. To get the user on the first subscribe to currentUser$ before reloadUser() is called this Subject should be a BehaviorSubject.

private _reloadUser = new BehaviorSubject<void>(0 as void);

currentUser$ = this._reloadUser.pipe(
  switchMap(() => {
    if (!this.jwtService.getToken() || this.jwtService.getAnonymousToken()) { 
      return of(null); 
    }
    console.warn('user.service.getcurrentuser -> load and CACHE it');
    return this.userService.loadCurrentUser();
  }),
  shareReplay(1),
  tap(user => console.log('user from CACHE:', user))
) 

reloadUser() {
  this._reloadUser.next();
}

With switchMap a call to reloadUser() will cancel any ongoing load request and send a new load request. You could also use exhaustMap instead of switchMap if a call to reloadUser() shouldn't reload the user if the user is currently being loaded.

shareReplay(1) is be better suited than publishReplay(1), refCount() in this case because it doesn't have the refCount feature by default. With publishReplay(1), refCount() a new subscriber to currentUser$ would trigger a load request if there aren't any subscribers present. shareReplay(1) will instead return the current user from the 'cache' in this case.

user_94232310

You are right, the line this.currentUser$ = this.userService.loadCurrentUser() is overwriting the property

Solution

Make a currentUser$ a Subject like below

currentUserSubject$ = new BehaviourSubject<User | null>(null);
currentUser$ = this.currentUserSubject$.asObservable();

Now we refactor your code to

getCurrentUser(): Observable<User> {
  if (!this.jwtService.getToken() || this.jwtService.getAnonymousToken()) { 
    return of(null); 
  }
    if (!this.currentUser$.value) {
      console.warn('user.service.getcurrentuser -> load and CACHE it once');
      return this.userService.loadCurrentUser().pipe(
        tap(user => this.currentUserSubject$.next(user)),
        // https://medium.com/angular-in-depth/fastest-way-to-cache-for-lazy-developers-angular-with-rxjs-444a198ed6a6
        publishReplay(1),
        refCount()
      );
    }
    console.log('user.service.getcurrentuser -> get from CACHE');
    return this.currentUser$;
 }

The clear cache function can be changed to

removeCachedUser(): void {
  this.currentUserSubject$.next(null);
}

Note the line tap(user => this.currentUserSubject$.next(user)), we tap into the observable stream and set the value of our subject using by calling the next() function on the Subject passing in the user

user_136801150

You can also use shareReplay with combination of takeUntil to remove cache.

https://stackoverflow.com/questions/47926240/when-should-i-use-publishreplay-vs-sharereplay/47926453#:~:text=publishReplay%20allows%20you%20to%20controls,to%20worry%20about%20unsubscribing%20etc.

private _destroyCache$ = new Subject<void>();

getCurrentUser(): Observable<User> {
    if (!this.jwtService.getToken() || this.jwtService.getAnonymousToken()) { return of(null); }
    if (!this.currentUser$) {
        console.warn('user.service.getcurrentuser -> load and CACHE it once');
        this.currentUser$ = this.userService.loadCurrentUser().pipe(
            takeUntil(this._destroyCache$),
            shareReplay(1),
        );
    }
    console.log('user.service.getcurrentuser -> get from CACHE');
    return this.currentUser$;
}

removeCachedUser(): void {
    this.currentUser$ = null;
    this._destroyCache$.next();
    this._destroyCache$.compleate();
    this._destroyCache$ = new Subject<void>();
}

And on the component you can have something like this:

ngOnInit() {
    loadData();
}

loadData() {
    this.currentUser$ = this.authService.getCurrentUser();
}

refreshData() {
    this.authService.removeCachedUser();
    loadData();
}
user_39557140
Related