我正在尝试学习如何在 Android 中使用 RxJava,但遇到了死胡同。我有以下数据源:
object DataSource {
enum class FetchStyle {
FETCH_SUCCESS,
FETCH_EMPTY,
FETCH_ERROR
}
var relay: BehaviorRelay<FetchStyle> = BehaviorRelay.createDefault(FetchStyle.FETCH_ERROR)
fun fetchData(): Observable<DataModel> {
return relay
.map { f -> loadData(f) }
}
private fun loadData(f: FetchStyle): DataModel {
Thread.sleep(5000)
return when (f) {
FetchStyle.FETCH_SUCCESS -> DataModel("Data Loaded")
FetchStyle.FETCH_EMPTY -> DataModel(null)
FetchStyle.FETCH_ERROR -> throw IllegalStateException("Error Fetching")
}
}
}
每当我更改 的值时,我都想触发下游更新relay
,但这不会发生。它在Activity
初始化时起作用,但在我更新值时不起作用。这是我的 ViewModel,我从中更新值:
class MainViewModel : ViewModel() {
val fetcher: Observable<UiStateModel> = DataSource.fetchData().replay(1).autoConnect()
.map { result -> UiStateModel.from(result) }
.onErrorReturn { exception -> UiStateModel.Error(exception) }
.startWith(UiStateModel.Loading())
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
fun loadSuccess() {
DataSource.relay.accept(DataSource.FetchStyle.FETCH_SUCCESS)
}
fun loadEmpty() {
DataSource.relay.accept(DataSource.FetchStyle.FETCH_EMPTY)
}
fun loadError() {
DataSource.relay.accept(DataSource.FetchStyle.FETCH_ERROR)
}
}
这是执行订阅的代码Activity
:
model.fetcher
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
uiState -> mainPresenter.loadView(uiState)
})