如何处理先前的请求并停止观察先前的LiveData

Question

我正在尝试在具有searchView的地方实现一个应用程序，当用户搜索某些内容时，这些应用程序将调用改造请求，该请求将返回可观察的数据。在myRepository中，将可观察对象转换为可流动对象，在myViewModel中，将可观察对象转换为LiveData，我将在myActivity中重新观察。

但是问题是，如果用户多次搜索并且真正快速（在获取先前结果之前），那么我想取消先前的请求，也不想观察先前的数据。

所以我正在使用的代码如下：

MainActivity

class MainActivity  : DaggerAppCompatActivity() {

 override fun onCreate(savedInstanceState: Bundle?) {
        observeRepos()
 }

 fun getReposFromServer(filter_search :String )    {
        mainViewModel.getReposFromServer(filter_search)    }

 private fun observeRepos() {
        mainViewModel.observeReposFromServer().observe(this, Observer { repos ->

            txtVwCount.setText("total item count is: "+repos?.total_count.toString())

        })

        mainViewModel.observeItemList().observe(this, Observer {

            if(!it.isNullOrEmpty())
                if(it.size>0) {
                    mAdapter.setReposInAdapter(it)
                    progressBar.visibility = View.GONE
                }
        })
    }


override fun onCreateOptionsMenu(menu: Menu?): Boolean {
        .....
        ......
         searchView.setOnQueryTextListener(object: SearchView.OnQueryTextListener{

          override fun onQueryTextSubmit(query: String?): Boolean {
                query?.let {
                            if(it.trim().length>0) {
                                clearOldCalls()
                                getReposFromServer(it,"","")      }
                            }
                        return false
            }

            override fun onQueryTextChange(newText: String?): Boolean {
                     return false;          }

        })
        return true
    }


fun clearOldCalls()   {
        mainViewModel.clearRetrofitCall()
        mAdapter.clearListInAdapter() //in my adapter I just make the list empty by assigning a new list to it
    }

}

MainViewModel：

class MainViewModel @Inject constructor() : ViewModel() {
    var liveGitResult = MediatorLiveData<GitResult>()
    val liveItemList =  MediatorLiveData<MutableList<ItemList>>()

     @set:Inject
    lateinit var mainRepository: MainRepository

    fun getReposFromServer(filter_search: String)    {
          val resultFromApiCall_flowable : Flowable<GitResult> =  mainRepository.fetchToDosFromServer(filter_search)
         lateinit var source: LiveData<GitResult>

         resultFromApiCall_flowable.let { 
                 source = LiveDataReactiveStreams.fromPublisher(it)

                  liveGitResult.addSource(source) { todos ->
                       liveGitResult.setValue(todos)
                       liveGitResult.removeSource(source)   }

            }

         var itemList_observable = resultFromApiCall_flowable.map {//it = gitResult
                    gitResult ->
                        var lst = mutableListOf<ItemList>()
                        gitResult.items.forEach {
                                                    lst.add(it)   }

                     lst
            }

        itemList_observable?.let{
                        var liveItemList  = LiveDataReactiveStreams.fromPublisher(itemList_observable)

                        this.liveItemList.addSource(liveItemList){ itemList ->
                        this.liveItemList.setValue(itemList)
                        this.liveItemList.removeSource(liveItemList)    }
            }


     fun observeReposFromServer(): LiveData<GitResult> {
        return liveGitResult
    }

    fun observeItemList(): LiveData<MutableList<ItemList>> {

        return liveItemList
    }

    fun clearRetrofitCall()
    {
        liveGitResult.value =null
        liveItemList.value = null
        mainRepository.clearDisposables()
    }

}

MainRepository：

class MainRepository @Inject constructor(mainApi: MainApi) {

    private val mainApi: MainApi
    private val disposables: CompositeDisposable = CompositeDisposable()


    init {
        this.mainApi = mainApi
    }

    fun fetchToDosFromServer(filter_search: String) : Observable<GitResult> {
 lateinit var  returnedData : Observable<GitResult>

      //mainApi.getAllRepo(filter_search) is a retrofit call which returns a   Flowable<GitResult>
             returnedData =    mainApi.getAllRepo(filter_search).subscribeOn( Schedulers.io())
                                                             .onErrorReturn(Function {throwable ->
                                                                Log.e( LOG_TAG, "Something went wrong" )
                                                                 null
                                                             })


              returnedData.subscribeOn(Schedulers.io())
                          .observeOn( AndroidSchedulers.mainThread())
                          .subscribe(object :Observer<GitResult>{
                                                override fun onSubscribe(d: Disposable) {
                                                    disposables.add(d)  }

                                                override fun onComplete() { }

                                                override fun onNext(t: GitResult) { }

                                                override fun onError(e: Throwable) { }

                                            })

            return returnedData.toFlowable(BackpressureStrategy.BUFFER);

    }


     fun clearDisposables(){
        if(disposables!=null){
            if(!disposables.isDisposed)
                disposables.clear()
                disposables.dispose()
            }
    }

}

现在，如果我运行该应用程序并真正快速搜索多次，然后再获得以前的结果-那么它的表现就很奇怪。

• 它在recyclerview中一一显示所有结果。以前的结果不应该被处理掉并且不会转换为liveData吗？
• 也未按顺序被称为
• 此外，可能不会使用来自不同Livedata来源的观察者同时显示recyclerview数据和textView数据

所以，我在哪里做错了？

也如何在单个语句中编写returnedData的代码，而不是在我正在执行的2条语句中编写它（[一个用于调用api，另一个用于覆盖订阅方法）。]]

我正在尝试在具有searchView的地方实现一个应用程序，当用户搜索某些内容时，这些应用程序将调用改造请求，该请求将返回可观察的数据。在myRepository中，我将...

Answer 1

对于类似的情况，我会保留一堆一次性物品，并在开始新的时处置最旧的订阅。我在其中运行OkHtttp请求，因此在处理时，它们被打断了。