typescript - Firebase e RxJS

Ho creato una funzione che ascolta gli eventi di valore:

getUnreadMessages() : Observable<any>{
    return new Observable(observer =>{
      firebase.database().ref('messages/' + this.userService.getUserID()).on("value", (snapshot : any) =>{

        let count = 0;
        ...some more logic here...

        observer.next(count);
        observer.complete();

      }, (error) =>{
        observer.error(error);
      })
    })
  }

Quando il database cambia, posso vedere che questa funzione viene eseguita, ma dove mi iscrivo ad essa in tutta l'applicazione non viene attivata. Qualche suggerimento?

Risposta n. 1

Contrassegnare il completamento osservabile è stata la causa del problema.