Reactivex Наблюдаемый поток пользовательского интерфейса блокировки

Question

Итак, я использую шаблон MVVM с Rx, дело в том, что когда я запускаю операцию, которая должна выполняться в фоновом потоке, мой поток пользовательского интерфейса блокируется, это моя модель представления:

class dataViewModel (application: Application): AndroidViewModel(application) {
        val dataRepository = DataRepository.getInstance(application.applicationContext)
        val listData = MutableLiveData<List<Data>>()
        var isLoading = MutableLiveData<Boolean>()

        fun loadData(){
            isLoading.value = true
            dataRepository.getData().subscribeOn(Schedulers.newThread())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeWith(object: DisposableObserver<List<Data>>(){
                        override fun onComplete() {
                            //Update UI
                            isLoading.value = false
                        }
                        override fun onNext(retrievedData: List<data>) {
                            listData.value = retrievedData
                        }
                        override fun onError(e: Throwable) {
                            //Handle error
                        }
                    })
        }
    }

Иэто метод репозитория:

fun getData(): Observable<List<Data>> {
    var dataList = query.search //this query may take a while and blocks the UI thread
    return Observable.just(dataList)
}

Я что-то упустил?

Answer 1

Ваш query.search происходит в потоке пользовательского интерфейса, потому что вы вызываете его до того, как ваши операторы Rx будут задействованы.Вы должны изменить getData метод на

fun getData(): Observable<List<Data>> {
    return Observable.fromCallable {
        query.search()
    }
}

В этом случае query.search будет вызываться в любом потоке, определенном в этой цепочке Rx (с использованием subscribeOn).

Answer 2

Вы просто вызываете свой тяжеловес за пределами цепочки RX, поэтому он блокируется.Попробуйте упаковать свою длительную операционную систему, например, в Observable.fromCallable, а все остальные части вашего блока (subscribeOn....) следуют.