使用Kotlin扩展函数将RxJava Observable转换为实时数据

Question

我一直在使用库在我的代码中使用很多RxJava Observables转换.所以我想在RxJava Observable中添加一个扩展函数来轻松转换它们.LiveDataLiveDataReactiveStreams.fromPublisher()LiveData

这些是我的扩展功能:

fun <T> Flowable<T>.toLiveData() :  LiveData<T> {
    return LiveDataReactiveStreams.fromPublisher(this)
}

fun <T> Observable<T>.toLiveData(backPressureStrategy: BackpressureStrategy) :  LiveData<T> {
    return LiveDataReactiveStreams.fromPublisher(this.toFlowable(backPressureStrategy))
}

fun <T> Single<T>.toLiveData() :  LiveData<T> {
    return LiveDataReactiveStreams.fromPublisher(this.toFlowable())
}

fun <T> Maybe<T>.toLiveData() :  LiveData<T> {
    return LiveDataReactiveStreams.fromPublisher(this.toFlowable())
}

fun <T> Completable.toLiveData() :  LiveData<T> {
    return LiveDataReactiveStreams.fromPublisher(this.toFlowable())
}

我的问题是:

1. 这是一个好主意吗？
2. 有没有更好的方法呢？
3. 这些扩展功能可以更好吗？

PS

我是Kotlin的新手,所以我问这些问题.任何有用的东西将不胜感激.非常感谢你.

Answer 1

我认为这是一个非常好的主意.一个示例好处LiveData是能够直接在数据绑定布局中使用它.让我们在您的视图模型中说:

val user: LiveData<User>

data class User(val firstName: String, val lastName: String)

在您的布局中,您可以User直接绑定属性:

android:text="${viewModel.user.firstName}"

您不能像这样在数据绑定中使用反应流.如果user是的话Flowable<User>,引用${viewModel.user.firstName}将无效.

此外,如果您的活动或片段调用,数据绑定将为您处理生命周期(仅在活动状态下观察更改并在更改发生时更新视图)ViewDataBinding.setLifecycleOwner(LifecycleOwner):

binding.setLifecycleOwner(this)

转换Completable到的那个LiveData<T>对我来说真的没有意义,因为它永远不会通知观察者什么,所以我只是摆脱它.

从反应流转换为实时数据时有一些注意事项(就像我想在旋转后恢复倒计时一样),但我认为它们与您提供的扩展功能无关,这些似乎可以完成它们的工作.这里要记住的问题是,当生命周期所有者从活动状态转为非活动状态时,PublisherLiveData取消对流的订阅,并且当状态更改为活动时,它将创建新订阅,这意味着在许多情况下重新启动流(我想这是流是"冷"),而你可能想要从旋转或其他配置更改后的位置恢复流.另一方面,如果流是"热"的,则在非活动状态期间忽略排放.我认为即使您直接使用反应流并手动处理生命周期,也必须解决此问题.但问题是,简单地将反应流转换LiveData为不足以解决这个问题.

将这些方法记录为不处理错误状态是很好的,错误状态必须在上游处理.或者,这可能是这些函数的改进之一 - 首先转换流以处理错误(例如,作为具有默认值的lambda参数).另一种可能性是利用Result(目前是实验性的)或类似的东西来封装成功或错误.

作为事后的想法,关于我上面写的这部分:

从反应流转换为实时数据时需要考虑一些因素,但我认为它们与您提供的扩展功能无关.

我仍然认为它拥有一般真实的,但我不知道,如果你真的想使用Single.toLiveData(),并Maybe.toLiveData()在实践中的大部分时间.由于Maybe并且Single正在建模一次性操作,因此当没有活动的观察者时可能最好不取消它,并且一旦有新的活动观察者就必须重新启动它.相反,发布到一些MutableLiveData并处理Single/ Maybein onCleared可能是有用的(我不确定是否可以封装在扩展函数中).它们仍然可能有一些我目前看不到的用途.

顺便说一句,你Flowable.toLiveData()已经在androidx.lifecycle:lifecycle-reactivestreams-ktx神器中了.

这留下了Observable.toLiveData(),我认为应该和它一样有用Flowable.

Answer 2

如果您想同时使用 LiveData 和 Rx，您的解决方案就很好。

如果您只想自动处理您的订阅，您可以像这样在 Disposable 上实现它：

private class LifecycleDisposable(obj: Disposable) :
        DefaultLifecycleObserver, Disposable by obj {
    override fun onStop(owner: LifecycleOwner) {
        if (!isDisposed) {
            dispose()
        }
    }
}

fun Disposable.attachToLifecycle(owner: LifecycleOwner) {
    owner.lifecycle.addObserver(LifecycleDisposable(this))
}

并称之为

Observable.just(1, 2, 3).subscribe().attachToLifecycle(this)

其中this引用任何 LifecycleOwner。