In rxjs, how to unsubscribe from "inner" "interval" observables using takeUntil or takeWhile

Question

[EDIT 11/5] After testing with some of the suggested solutions, I noticed some of my rxjs imports were wrong (I was importing from 'rxjs/internal' iso from 'rxjs'). So if you come across an error like this, that's something that you might have a look at.

There already exist a couple of questions on Stackoverflow related to unsubscribing from an observable, though none of them are very helpful to my specific problem.

In my ("chat") application, I can search for users by typing their name in a search box. After fetching the users, I want to display whether these users are online or not.

Here's my approach.

There's a BehaviorSubject for capturing the search term:

private term$ = new BehaviorSubject(null);

Whenever there's a new value on the term$ observable, I start looking for new users:

users$: Observable<User[]> = this.term$.pipe(
    filter(term => !!term),
    switchMap(term => this.usersService.searchUsers(term))
);

But I also want to periodically check whether these users are "online", hence I change above Observable like so (note that last line):

users$: Observable<any[]> = this.term$.pipe(
    filter(term => !!term),
    switchMap(term => this.usersService.searchUsers(term)),
    switchMap(users => zip(...users.map(user => this.checkLoggedInF(user))))
);

For all users I create an "Interval" observable. checkedLoggedInF is a function that checks every 5 seconds with the server whether the given user is online:

checkLoggedInF = user => {
    return interval(5000).pipe(
      switchMap(() => this.usersService.isLoggedIn(user.loginnaam).pipe(
        map(loggedIn => ({...user, loggedIn}))
      ))
    );
}

Now the problem is that whenever there is a new search term (on the term$ observable), the "checkLoggedIn interval" observables should be unsubscribed to. I have tried using a takeUntil operator in both the "checkLoggedIn interval" observable and the parent "users$" observable, but to no avail. Also using a takeWhile operator was ineffective.

takeUntil should do what you want. I wrote a user-land operator to do this sort of thing: github.com/cartant/rxjs-etc/blob/master/source/operators/… You might also want to read the Strongbrew article that's referenced in a comment at the top of that file. — cartant

Artyom Amiryan Artyom Amiryan · Accepted Answer · 2018-11-02T16:29:03

try to use .tap() before switchMap where you are calling checkLoggedInF, and save your observable from checkLoggedInF in a property

checkLoggedInF = user => {
    this.checkLoggedInF$ = interval(5000).pipe(
      switchMap(() => this.usersService.isLoggedIn(user.loginnaam).pipe(
        map(loggedIn => ({...user, loggedIn}))
      ))
    );
    return this.checkLoggedInF$;
}

users$: Observable<any[]> = this.term$.pipe(
    filter(term => !!term),
    switchMap(term => this.usersService.searchUsers(term)),
    tap(() => { this.checkLoggedInF$ && this.checkLoggedInF$.unsubscribe() })
    switchMap(users => zip(...users.map(user => this.checkLoggedInF(user))))
);

or another option you can check and unsubscribe() in your checkLoggedInF too

checkLoggedInF = user => {
   this.checkLoggedInF$ && this.checkLoggedInF$.unsubscribe() 
   this.checkLoggedInF$ = interval(5000).pipe(
     switchMap(() => this.usersService.isLoggedIn(user.loginnaam).pipe(
       map(loggedIn => ({...user, loggedIn}))
     ))
   );
   return this.checkLoggedInF$;
}

In rxjs, how to unsubscribe from "inner" "interval" observables using takeUntil or takeWhile

2 Answers