RxJava для обработки ошибок данных в реальном времени

Question

Итак, в моем проекте Android я использую RxJava в слое хранилища, а затем преобразовываю его в живые данные в представлении Model, но у меня возникает исключение, когда мой поток выдает ошибку. Это как-то связано с этим замечанием из Документации Android :

fromPublisher

LiveData<T>fromPublisher (Publisher<T> publisher)

Создает наблюдаемый поток LiveData от ReactiveStreams издателя.

Когда LiveData становится активным, он подписывается на выбросы от издателя.

Когда LiveData становится неактивным, подписка очищается. LiveData содержит последнее значение, выданное издателем, когда LiveData был активным.

Следовательно, в случае горячей наблюдаемой RxJava, когда добавляется новый LiveData Observer, он будет автоматически уведомлять с последним значением в LiveData, которое может быть последнее значение, выданное издателем.

Обратите внимание, что LiveData НЕ обрабатывает ошибки и ожидает, что ошибки обрабатываются как состояния в данных, которые хранятся. В случае, если издатель выпустил ошибку, она будет передана в основной поток, и приложение вылетит.

Как я могу обработать поток ошибок PS: я использую расширение Kotlin для преобразования в живые данные

fun <T> Observable<T>.toLiveData(backPressureStrategy: BackpressureStrategy =
                                                BackpressureStrategy.LATEST) :  LiveData<T> {
Log.d("RxJava",this.toString())
return LiveDataReactiveStreams.fromPublisher(this.toFlowable(backPressureStrategy))

Answer 1

Как новичок, я не уверен в этом решении. Но это работает. Так что, если это плохо, пожалуйста, прокомментируйте, чтобы сказать, почему и предоставьте альтернативу.

Используйте это:

data class DataWithStates<T>(
    val data: T? = null,
    val states: Throwable? = null
)

Образец для ViewModel:

fun getUsers(): LiveData<DataWithStates<List<User>>> {
    return LiveDataReactiveStreams.fromPublisher(
        userRepository
            .getAll()
            .map { lstUser -> DataWithStates(lstUser) }
            .onErrorReturn { ex -> DataWithStates(states = ex) }
            .toFlowable(BackpressureStrategy.LATEST))
}

Может быть, лучше переместить .map и .onErrorReturn в хранилище.

Редактировать: в качестве альтернативы вы можете использовать лямбду для прослушивания ошибок.

fun getUsers(onError: () -> Unit = {}): LiveData<List<User>> {
    return LiveDataReactiveStreams.fromPublisher(
        userRepository
            .getAll()
            .doOnError { onError() }
            .toFlowable(BackpressureStrategy.LATEST))
}

Answer 2

, если вы используете живые данные (компоненты Android Architecure). Лучше использовать его предопределенный оберсервер (который находится в компонентах Architecutural). Иначе полностью вам нужно использовать только RX android clases