使用 RxJava 或 kotlin 协程的 Android ViewState答案

【问题标题】：Android ViewState using RxJava or kotlin coroutines使用 RxJava 或 kotlin 协程的 Android ViewState
【发布时间】：2018-07-19 21:42:09
【问题描述】：

我正在尝试学习如何在 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)
            })

【问题讨论】：

删除.replay(1).autoConnect()只要你使用的是BehaviorRelay，它具有.replay(1).autoConnect()开箱即用的功能。
这个改变好像不起作用。它实际上使我的 ViewModel 重新开始旋转获取。我试图阻止。每当中继获得新值时，我希望触发 fetchData()。
loadData 中的 Thread.sleep 很可能在主线程上执行（假设在主线程上调用了 loadX），从而在 5 秒内阻止任何 UI 更新。否则我在显示的代码中看不到任何问题。也许是时候开始在各个地方添加doOnNext { prinln(it) } 以查看数据丢失的位置。
@akarnokd 你是对的。我放在地图函数中的日志消息：“线程内部地图：main”。你怎么能把它从主线程中移开？
relay.observeOn.map

标签： android rx-java2 kotlinx.coroutines

【解决方案1】：

最终改用 kotlin 协程，因为我无法重新订阅 ConnectableObservable 并开始新的提取。

这里是任何感兴趣的人的代码。

主持人：

class MainPresenter(val view: MainView) {

    private lateinit var subscription: SubscriptionReceiveChannel<UiStateModel>

    fun loadSuccess(model: MainViewModel) {
        model.loadStyle(DataSource.FetchStyle.FETCH_SUCCESS)
    }

    fun loadError(model: MainViewModel) {
        model.loadStyle(DataSource.FetchStyle.FETCH_ERROR)
    }

    fun loadEmpty(model: MainViewModel) {
        model.loadStyle(DataSource.FetchStyle.FETCH_EMPTY)
    }

    suspend fun subscribe(model: MainViewModel) {
        subscription = model.connect()
        subscription.subscribe { loadView(it) }
    }

    private fun loadView(uiState: UiStateModel) {
        when(uiState) {
            is Loading -> view.isLoading()
            is Error -> view.isError(uiState.exception.localizedMessage)
            is Success -> when {
                uiState.result != null -> view.isSuccess(uiState.result)
                else -> view.isEmpty()
            }
        }
    }

    fun unSubscribe() {
        subscription.close()
    }
}

inline suspend fun <E> SubscriptionReceiveChannel<E>.subscribe(action: (E) -> Unit) = consumeEach { action(it) }

观点：

...
override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        launch(UI) {
            mainPresenter.subscribe(model)
        }

        btn_load_success.setOnClickListener {
            mainPresenter.loadSuccess(model)
        }

        btn_load_error.setOnClickListener {
            mainPresenter.loadError(model)
        }

        btn_load_empty.setOnClickListener {
            mainPresenter.loadEmpty(model)
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        Log.d("View", "onDestroy()")
        mainPresenter.unSubscribe()
    }
...

型号：

class MainViewModel : ViewModel() {

    val TAG = this.javaClass.simpleName

    private val stateChangeChannel = ConflatedBroadcastChannel<UiStateModel>()

    init {
        /** When the model is initialized we immediately start fetching data */
        fetchData()
    }

    override fun onCleared() {
        super.onCleared()
        Log.d(TAG, "onCleared() called")
        stateChangeChannel.close()
    }

    fun connect(): SubscriptionReceiveChannel<UiStateModel> {
        return stateChangeChannel.openSubscription()
    }

    fun fetchData() = async {
        stateChangeChannel.send(UiStateModel.Loading())
        try {
            val state = DataSource.loadData().await()
            stateChangeChannel.send(UiStateModel.from(state))

        } catch (e: Exception) {
            Log.e("MainModel", "Exception happened when sending new state to channel: ${e.cause}")
        }
    }

    internal fun loadStyle(style: DataSource.FetchStyle) {
        DataSource.style = style
        fetchData()
    }
}

这里是a link to the project on github。

【讨论】：