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

Question

我们的应用具有用于搜索地点的 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 ？

提前致谢！

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