Convert LiveData to RxJava observable

Question

I have 2 views: EditText with amount (which should not be empty) and agreement checkbox (which should be checked). I also have 2 MutableLive data variables which represent states of this views inside my ViewModel.

I want to combine this 2 variables in Observables and use Observable.combineLattest to enable/disable my "Send" button.

I have found library which called android.arch.lifecycle:reactivestreams and converted my LiveData to Publishers, but I cannot use them in Observable.combineLattest because org.reactivestreams because Publisher is org.reactivestreams interface and Observable.combineLattest accept observable source.

I read some articles but they all refer to this library.

Currently I have code like this:

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
                })
    }

Is anyone know good workaround to convert LiveData to Observable. Thanks in advance.

Answer 1

An alternative solution might be (I needed this because WorkManager only returns 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()
                }
            }
        }
    }