AsyncSubject可以处理RxJava中的SingleLiveEvent情况吗？

Question

问题

一次性事件的无功编程模式的问题是，在发生最初的一次性事件后，它们可能会重新发送给订户。

对于LiveData，the SingleLiveEvent提供使用EventObserver的解决方案，也可以应用于Kotlin Flow。

问题

是否可以创建可观察到的EventObserver来处理RxJava中SingleLiveEvent的情况？主要问题似乎是在调用AsyncSubject之后，是否有办法手动“重新打开” AsyncSubject以重新发送数据？

潜在的解决方案

[AsyncSubject似乎是RxJava的潜在解决方案，无需创建onComplete，因为AsyncSubject指出它将仅在序列完成时才发布。

实施-加载状态示例

从ViewModel方法EventObserver发出加载布尔值，并将视图效果状态发送给视图，在这种情况下为片段。加载布尔值在片段初始化和ViewModel通过documentation发送initFeed并在成功或错误尝试中均以true完成时按预期工作。

但是，例如重新刷写刷新相同的onNext方法时，重新发送值的尝试将失败。似乎在为同一对象调用onComplete之后无法使用initFeed。

SomeViewEffect.kt

onNext
SomeViewModel.kt

onComplete
SomeFragment.kt

data class _FeedViewEffect(
    val _isLoading: AsyncSubject<Boolean> = AsyncSubject.create(),
)

data class FeedViewEffect(private val _viewEffect: _FeedViewEffect) {
    val isLoading: AsyncSubject<Boolean> = _viewEffect._isLoading
}

问题一次性事件的反应式编程模式的问题在于，可能在最初的一次性事件发生后将它们重新发送给订户。对于LiveData，...

Answer 1

尚不清楚为什么您需要仅发出最后一个事件的AsyncSubject。您是否尝试过在这种情况下使用

private fun initFeed(toRetry: Boolean) {
        val disposable = feedRepository.initFeed(pagedListBoundaryCallback(toRetry))
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { results ->
                when (results.status) {
                    LOADING -> {
                        Log.v(LOG_TAG, "initFeed ${LOADING.name}")
                        _viewEffect._isLoading.onNext(true)
                    }
                    SUCCESS -> {
                        Log.v(LOG_TAG, "initFeed ${SUCCESS.name}")
                        _viewEffect._isLoading.onNext(false)
                        _viewEffect._isLoading.onComplete()
                        _viewState._feed.onNext(results.data)
                    }
                    ERROR -> {
                        Log.v(LOG_TAG, "initFeed ${ERROR.name}")
                        _viewEffect._isLoading.onNext(false)
                        _viewEffect._isLoading.onComplete()
                        _viewEffect._isError.onNext(true)
                    }
                }
            }
        disposables.add(disposable)
    }

或

private fun initViewEffects() {
        val isLoadingDisposable = viewModel.viewEffect.isLoading
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError { Log.v(LOG_TAG, "Error loading isLoading") }
            .subscribe { isLoading ->
                if (isLoading) progressBar.visibility = VISIBLE
                else {
                    progressBar.visibility = GONE
                    swipeToRefresh.isRefreshing = false
                }
            }
        compositeDisposable.addAll(isLoadingDisposable, isErrorDisposable)
    }

处理器？