Search code examples
androidkotlinrx-java2rx-kotlin2

RxJava2 onNext() called multiple times?

I have a method that returns an Observable like this:

open fun get(): Observable<Response> {

    return if (condition)
        getDataFromApi()
    else
        getDataFromDb()

}

and is subscribed as followed:

                get()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(/*a object of class extending DefaultObserver*/)

I am facing this issue when getDataFromDb() is called and after a while getDataFromApi() is called as per condition. For first call it works fine but on second call onNext is called more then one time with the old data response from getDataFromDb(). Please let me know what I am doing wrong. I am a bit new to RxJava.

Solution

  • If getDataFromDb() is emitting items and getDataFromApi() is called, the first method will continue to emit till finish. You should unsubscribe from the stream if it is not needed anymore and also add condition in your source in order to stop emitting if observable is unsubscribed.

    Also keep in mind that the functions inside an observable do not run till the stream is subscribed.