RxJava2 onNext() вызывается несколько раз?

Question

У меня есть метод, который возвращает Observable, как это:

open fun get(): Observable<Response> {

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

}

и подписывается следующим образом:

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

Я сталкиваюсь с этой проблемой, когда getDataFromDb() вызывается и через некоторое время getDataFromApi() вызывается согласно условию. Для первого вызова он работает нормально, но при втором вызове onNext вызывается более одного раза со старым ответом на данные от getDataFromDb(). Пожалуйста, дайте мне знать, что я делаю неправильно. Я немного новичок в RxJava.

Answer 1

Если getDataFromDb() отправляет элементы и вызывается getDataFromApi(), первый метод будет продолжать излучать до конца. Вы должны отписаться от потока, если он больше не нужен, а также добавить условие в ваш источник, чтобы прекратить излучение, если наблюдаемое отписано.

Также имейте в виду, что функции внутри наблюдаемой не выполняются до тех пор, пока поток не будет подписан.