Using Rx debounce operator in Filter.performFiltering() in Adapter for Place API

Question

Our app has AutoCompleteTextView used to search places, with Google Places API.

I WANT TO ADD debounce RxJava OPERATOR TO PLACE SEARCH. I am not very good at RxJava.

Adapter has getFilter() implemented and Filter has performFiltering() overiden that receives the constraint/query as string.
Filter currently looks like this:

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

And fetchAutoCompletePredictions() looks like this:

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

I tried to use the PublichSubject (PublishRelay from JakeWharton library) but I am still not confident of fixing it with this as EVENT happens here through this call (getting constraint), and the same method should also return FIlterResult. This means observer should be placed in performFiltering() as well. This means for every entry of letter, this method hits and so multiple observers.
And also, I should cal subject.onNext() in the same method as the new search quesry (constraint) is only known here.

How can I use debounce operator in this case to make the whole process sync, return FIlterResults at the end of it ?

Thanks in advance!

Answer 1

I got rid of the filter. Instead of letting filter callbacks gets notified about new search string, I added my own publishSubject that observes the new search string and notifies the observers.

Own observer (with publicSubject instead of filter callback) allowed me to insert debounce operator in the stream.

Added this in adapter:

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

Then these two in 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()
}

Have set the new search string to subject from my presenter:

adapter.setQuery(currText)