将 LiveData 转换为 RxJava 可观察对象答案

【问题标题】：Convert LiveData to RxJava observable将 LiveData 转换为 RxJava 可观察对象
【发布时间】：2019-08-12 14:50:07
【问题描述】：

我有 2 个视图：带有金额的 EditText（不应为空）和协议复选框（应选中）。我还有 2 个 MutableLive 数据变量，它们代表我的 ViewModel 中此视图的状态。

我想在 Observables 中组合这 2 个变量并使用 Observable.combineLattest 来启用/禁用我的“发送”按钮。

我找到了一个名为 android.arch.lifecycle:reactivestreams 的库并将我的 LiveData 转换为 Publishers，但我不能在 Observable.combineLattest 中使用它们，因为 org.reactivestreams 因为 Publisher 是 org.reactivestreams 接口并且 Observable.combineLattest 接受可观察源.

我读了一些文章，但它们都引用了这个库。

目前我有这样的代码：

override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        super.onViewCreated(view, savedInstanceState)

        ...

        dispossable = Observable.combineLatest(
                LiveDataReactiveStreams.toPublisher(this, vm.amount),
                LiveDataReactiveStreams.toPublisher(this, vm.isAgreementChecked),
                BiFunction<String, Boolean, Boolean> { amount, isChecked ->
                    amount.isNotEmpty() && isChecked
                })
    }

有没有人知道将 LiveData 转换为 Observable 的好方法。提前致谢。

【问题讨论】：

Flowable. combineLatest ?
任何 ObservableSource
您需要Publisher。 Flowable 是 Publisher
如果你绝对需要一个 observable，你可以使用Observable.fromPublisher(LiveDataReactiveStreams.toPublisher(...))。
好的，非常感谢。我（出于习惯）使用 Observable 搜索解决方案，但不记得 Flowable 和发布者。

标签： android rx-java android-livedata

【解决方案1】：

正如 Blackbelt 所说，我可以使用 LiveData 中的 Flowable，而不是使用 Observable 进行 combineLatest（以及其他运算符，如 Zip、Debounce 等）：

LiveDataReactiveStreams.toPublisher(/*lifecycle*/, /*observable Field*/).

所以我目前的解决方案是这样的：

disposable = Flowable.combineLatest(
    LiveDataReactiveStreams.toPublisher(this, vm.amount),
    LiveDataReactiveStreams.toPublisher(this, vm.isAgreementChecked),
    BiFunction<String, Boolean, Boolean> { amount, isChecked ->
        amount.isNotEmpty() && isChecked
    }).subscribe { isDataValid ->
        vm.setIsDataValid(isDataValid)
    }

再次感谢:)

【讨论】：

【解决方案2】：

另一种解决方案可能是（我需要这个，因为 WorkManager 只返回 LiveData）：

fun getWorkData(): Flowable<List<WorkInfo>> {
    val workDataLiveData = WorkManager.getInstance().getWorkInfosByTagLiveData("TAG")
    Flowable.create({emitter -> 
        val observer = Observer> { emitter.onNext(it) }
        val disposable = disposeInUiThread { workDataLiveData.removeObserver(observer) }
        emitter.setDisposable(disposable)
        workDataLiveData.observeForever(observer)
    }, BackpressureStrategy.LATEST)
}

private fun disposeInUiThread(action: Action): Disposable {
        return Disposables.fromAction {
            if (Looper.getMainLooper() == Looper.myLooper()) {
                action.run()
            } else {
                val inner = AndroidSchedulers.mainThread().createWorker()
                inner.schedule {
                    try {
                        action.run()
                    } catch (e: Exception) {
                        Timber.e(e, "Could not unregister receiver in UI Thread")
                    }

                    inner.dispose()
                }
            }
        }
    }

【讨论】：