在Place API的适配器中的Filter.performFiltering()中使用Rx反跳运算符

Question

我们的应用程序有AutoCompleteTextView用于搜索地点，与谷歌地点API。

我想在PLACE SEARCH中添加debounce RxJava操作符。我不太擅长RxJava。

适配器实现了getFilter（），而过滤器具有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()
            }
        }
    }
}

和fetchAutoCompletePreVII（）如下所示：

 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（）中。这意味着对于每个字母条目，此方法都会命中，因此会有多个观察者。
而且，我应该以相同的方法cal subject.onNext（）作为新的搜索查询（约束）只在这里知道。

在这种情况下，如何使用防抖运算符使整个过程同步，在结束时返回FIlter结果？

提前致谢！

Answer 1

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

自己的观察者（与公共主题而不是过滤器回调）允许我在流中插入防抖运算符。

在适配器中添加了这个：

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

然后这两个在适配器中：

    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)