typescript - Firebase dan RxJS

Saya membuat sebuah fungsi yang mendengarkan event nilai:

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);
      })
    })
  }

Ketika database berubah, saya bisa melihat bahwa fungsi ini dijalankan, namun tempat saya berlangganan padanya di seluruh aplikasi tidak aktif. Ada saran?

Jawaban #1

Menandai yang dapat diamati sebagai selesai adalah penyebab masalahnya.