rxjs - Cancel previous inner Observables when a new higher-order Observable is emitted -

consider following code:

this.msgservice.getuserchatlist()     .do( (list) => {        this.unread = false;       console.log(list);      } )     .mergemap( chats => observable.from(chats) )     .mergemap( chat => this.msgservice.getlastmessage(chat['id']).map( lastmessage => this.containsunreadmessages(lastmessage, chat['lastpresence']) ) )     .filter( state => state === true )     .subscribe( (unread) => {       this.unread = true;       console.log('result ', res);     } ); 

getuserchatlist(): - emits element each time 1 of chat changes - element raw array containing chats meta-data - never completes

getlastmessage(): - observable never completes

in second mergemap calling function getlastmessage(). need observe observable only until new item emitted getuserchatlist() otherwise multiple observations on last message of same chat.

illustration :

1. getuserchatlist emits : [chatmetaa:{}, chatmetab:{}]
2. code go through getlastmessage , start observe lastmessage of chata , chatb
3. one of chat change new item emitted getuserchatlist containing new version of meta-data of chats: [chatmetaa:{}, chatmetab:{}]
4. code go through getlastmessage , start observe lastmessage of chata , chatb. observe twice last message of chata , chatb

and go on , on...

my question is, how cancel observation on getlastmessage() once new item emitted getuserchatlist()? tried using switch couldn't manage make work

solution indeed use switchmap:

this.msgservice.getuserchatlist()     .do( () => { this.unread = false } )     .switchmap(       chats => observable.from(chats)         .mergemap( chat => this.msgservice.getlastmessage(chat['id'])         .map( lastmessage => this.containsunreadmessages(lastmessage, chat['lastpresence']) ) )     )     .filter( state => state === true )     .subscribe( (unread) => {       this.unread = true;       console.log('result ', res);     } );