在 Adapter for Place API 的 Filter.performFiltering() 中使用 Rx debounce 运算符

Posted 2023-04-19

【中文标题】在 Adapter for Place API 的 Filter.performFiltering() 中使用 Rx debounce 运算符

【英文标题】：Using Rx debounce operator in Filter.performFiltering() in Adapter for Place API

【发布时间】：2019-12-07 14:31:16

【问题描述】：

我们的应用具有用于搜索地点的 AutoCompleteTextView，以及 Google Places API。

我想添加 debounce RxJava OPERATOR 来进行搜索。我不太擅长 RxJava。

Adapter 已实现 getFilter()，Filter 已覆盖 performFiltering()，以字符串形式接收约束/查询。 过滤器目前看起来像这样：

private inner class AutocompleteFilter : Filter() 

    @WorkerThread
    override fun performFiltering(constraint: CharSequence?): Filter.FilterResults 
        val results = Filter.FilterResults()
        val search = constraint?.toString() ?: ""

        var found = emptyList<PlaceData>()

        // quick reply on empty search
        val bounds = latLngBounds
        if (!TextUtils.isEmpty(search) && bounds != null) 
            found = placesApiHelper.fetchAutocompletePredictions(search, bounds)
                    .timeout(1000L, TimeUnit.MILLISECONDS)
                    .toList()
                    .doOnError 
                        searchResultSubject.onNext(false)
                        logTo(log).warning("Timeout or error happens. %s", it)
                    
                    .doOnSuccess 
                        searchResultSubject.onNext(true)
                    
                    .onErrorReturn  emptyList() 
                    .blockingGet()
        

        results.values = found
        results.count = found.size

        return results
    

    @MainThread
    override fun publishResults(constraint: CharSequence?, results: Filter.FilterResults) 
        logTo(log).info("publishing results. Found: %s items", results.count)

        data.clear()

        val list = results.values as? List<*>
        list?.let 
            val values = it.filterIsInstance<PlaceData>()
            if (values.isNotEmpty()) 
                data.addAll(values)
                notifyDataSetChanged()
            
        
    

而 fetchAutoCompletePredictions() 看起来像这样：

 private fun fetchAutocompletePredictions(query: String, latLngBounds: LatLngBounds): Observable<PlaceData> 
    Timber.d("places api request: %s (total: %s, cached: %s)", query, PlacesApiCache.CALLS.incrementAndGet(), PlacesApiCache.HITS.get())

    val rectangularBounds = toRectangularBounds(latLngBounds)

    val request = FindAutocompletePredictionsRequest.builder()
            .setLocationBias(rectangularBounds)
            .setQuery(query)
            .build()

    return Observable.create 
        placesClient.findAutocompletePredictions(request).addOnSuccessListener  response ->

            val predictions = response.autocompletePredictions
            val results = mutableListOf<PlaceData>()

            for (i in 0 until Math.min(predictions.size, MAX_RESULTS)) 
                results.add(placeDataFrom(predictions[i]))
            

            PlacesApiCache.cacheData(query, results)

            for (result in results) 
                it.onNext(result)
            

            it.onComplete()
        .addOnFailureListener  exception ->
            if (exception is ApiException) 
                Timber.e("Place not found: %s", exception.getStatusCode())
            
            it.onError(exception)
        
    

我尝试使用 PublichSubject（来自 JakeWharton 库的 PublishRelay），但我仍然没有信心修复它，因为 EVENT 通过此调用（获取约束）在这里发生，并且相同的方法也应该返回 FIlterResult。这意味着观察者也应该放在 performFiltering() 中。这意味着对于每个字母条目，此方法都会命中多个观察者。 而且，我应该使用与新搜索查询（约束）相同的方法来调用 subject.onNext() 仅在此处知道。

在这种情况下如何使用 debounce 运算符使整个过程同步，最后返回 FIlterResults ？

提前致谢！

【问题讨论】：

我希望 Places API 构建器有助于消除抖动并为应用程序做这些事情。它不这样做。当然，他们为什么要鼓励。让应用调用API，增加成本。 :)

【参考方案1】：

我去掉了过滤器。我没有让过滤器回调收到有关新搜索字符串的通知，而是添加了自己的 publishSubject 来观察新的搜索字符串并通知观察者。

自己的观察者（使用 publicSubject 而不是过滤器回调）允许我在流中插入去抖动运算符。

在适配器中添加了这个：

private var queryPlacesSubject = PublishSubject.create<String>()

init 
    subscribeForPlacePredictions()


private fun subscribeForPlacePredictions() 
    queryPlacesSubject
            .debounce(DEBOUNCE_TIME_MILLIS, TimeUnit.MILLISECONDS)
            .distinctUntilChanged()
            .observeOn(Schedulers.newThread())
            .doOnNext  findPlacePredictions(it) 
            .observeOn(androidSchedulers.mainThread())
            .doOnNext notifyThePlacesChanged()
            .subscribeOn(Schedulers.io())
            .subscribe()


fun setQuery(query: String) 
    queryPlacesSubject.onNext(query)

那么这两个在adapter中：

    private fun findPlacePredictions(query: String?) 
    val search = query ?: ""

    // save entered text by user for highlight operations
    highlight = search

    // we should be in background thread already
    var found = emptyList<PlaceData>()

    // quick reply on empty search
    val bounds = latLngBounds
    if (!TextUtils.isEmpty(search) && bounds != null) 
        found = placesApiHelper.requestAutocomplete(search, bounds)
                .timeout(runtime.timeoutAutocompleteSearch(), TimeUnit.MILLISECONDS)
                .toList()
                .doOnError 
                    searchResultSubject.onNext(false)
                    logTo(log).warning("Timeout or error happens. %s", it)
                
                .doOnSuccess 
                    searchResultSubject.onNext(true)
                
                .onErrorReturn  emptyList() 
                .blockingGet()
    

    data.clear()
    data.addAll(found)


private fun notifyThePlacesChanged() 
    notifyDataSetChanged()

已将新的搜索字符串设置为我的演示者的主题：

adapter.setQuery(currText)