synchronous calls with rxjava Android

Question

My app has a SearchView. When the user types in the SearchView the onQueryTextChange passes the query to the presenter and then it calls the API. I am using Retrofit and RxJava for the calls. The calls return a json file with the words containing what the user typed so far. The problem is that, if the user is fast to type letters and the network is slow sometimes the SearchView doesn't show the results based on all the typed letters but maybe up to the second last because the last call was quicker to get the results compared to the second last.

Example: the user start typing:

"cou" -> make a call to the API (first call after 3 letters) -> start returnin values

"n" -> make a call -> start returning values

"t" -> make a call -> start returning values

"r" -> make a call (the connection is slow)

"y" -> make a call -> start returning values

-> "r" get the results finally and the returns them

public Observable<List<MyModel>> getValues(String query) {
    return Observable.defer(() -> mNetworkService.getAPI()
            .getValues(query)
            .retry(2)
            .onErrorReturn(e -> new ArrayList<>()));
}

The call is very simple and whenever I get an error I don't want to display anything.

Is there a way to solve that? Or maybe this is not the case to use reactive programming?

EDIT: Just to make more clear, the flow is the following:

1. Activity that uses a custom search view (https://github.com/Mauker1/MaterialSearchView)

2. the custom searchview has a listener when the user starts typing. Once the user starts typing the activity calls the Presenter.

3. the presenter will subscribe an observable returned by the interactor:

presenter:

addSubscription(mInteractor.getValues(query)
            .observeOn(mMainScheduler)
            .subscribeOn(mIoScheduler)
            .subscribe(data -> {
                getMvpView().showValues(data);
            }, e -> {
                Log.e(TAG, e.getMessage());
            }));

interactor:

public Observable<List<MyModel>> getValues(String query) {
    return Observable.defer(() -> mNetworkService.getAPI()
            .getValues(query)
            .debounce(2, TimeUnit.SECONDS)
            .retry(2)
            .onErrorReturn(e -> new ArrayList<>()));

So now either I change the custom search view in a 'normal' searchview and then use RxBinding or maybe I should use an handler or something like that (but still struggling how to fit it in my architecture)

Answer 1

Firstly make your Searchview as Observable so that you can apply Rx operators. To convert searchview into Observable

public static Observable<String> fromview(SearchView searchView) {
final PublishSubject<String> subject = PublishSubject.create();

searchView.setOnQueryTextListener(new SearchView.OnQueryTextListener() {
    @Override
    public boolean onQueryTextSubmit(String s) {
        subject.onComplete();
        searchView.clearFocus(); //if you want to close keyboard
        return false;
    }

    @Override
    public boolean onQueryTextChange(String text) {
        subject.onNext(text);
        return false;
    }
});

return subject;

}

private void observeSearchView() {

disposable = RxSearchObservable.fromview(binding.svTweet)
        .debounce(300, TimeUnit.MILLISECONDS)
        .filter(text -> !text.isEmpty() && text.length() >= 3)
        .map(text -> text.toLowerCase().trim())
        .distinctUntilChanged()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe()

}

You can apply filter, condition RxJava debounce() operator to delay taking any action until the user pauses briefly.

Use of distinctUntilChanged() ensures that the user can search for the same thing twice, but not immediately back to back

The filter operator is used to filter the unwanted string like the empty string in this case to avoid the unnecessary network call.

Handling searchview withRXJava

Answer 2

You're in luck there's an operator for that called debounce

Observable.defer(() -> mNetworkService.getAPI()
            .getValues(query)
            .debounce(3, TimeUnit.SECONDS)
            .retry(2)
            .onErrorReturn(e -> new ArrayList<>()));

What debounce does is wait N time units for more results prior to continuing. Say for example the network takes 2 seconds to return and you flood it with request after request, debounce will wait for 3 seconds of no results and then return the last result. Think of it as dropping everything but the one before N time of inactivity.

This solve your problem but will still flood the network, ideally you would use the excellent RxBinding library do the defer prior to making the request something like:

RxTextView.textChanges(searchView)
.debounce(3, TimeUnit.SECONDS)
.map(input->mNetworkService.getAPI().getValues(input.queryText().toString()))
.retry(2)
.onErrorReturn(e -> new ArrayList<>()))

With the current setup it will wait 3 seconds after a user types something and only then make the network call. If instead they start typing something new, the first pending search request gets dropped.

Edit: changed to RxTextView.textChanges(textview) based on OP not using an android SearchView widget

Answer 3

Extending on what @MikeN said, if you want to only use the results of the LAST input, you should use switchMap() (which is flatMapLatest() in some other Rx implementations).

Answer 4

I solved the flooding issue without using RxBinding and I want to post my solution just in case someone else needs it. So whenever the onTextChanged is called I check, first of all, if the size is > 2 and if it is connected to the network (boolean updated by a BroadcastReceiver). Then I create message to be sent has delayed and I delete all the other messages in the queue. This means that I will execute only the queries that are not within the specified delay:

@Override
    public void onTextChanged(CharSequence s, int start, int before, int count) {

        if (TextUtils.getTrimmedLength(s) > 2 && isConnected) {
            mHandler.removeMessages(QUERY_MESSAGE);
            Message message = Message.obtain(mHandler, QUERY_MESSAGE, s.toString().trim());
            mHandler.sendMessageDelayed(message, MESSAGE_DELAY_MILLIS);
        }
    }

Then the Handler:

private Handler mHandler = new Handler() {
    @Override
    public void handleMessage(Message msg) {
        if (msg.what == QUERY_MESSAGE) {
            String query = (String)msg.obj;
            mPresenter.getValues(query);
        }
    }
};

Answer 5

1. Add rxbinding dependency to gradle implementation "com.jakewharton.rxbinding2:rxbinding-kotlin:2.1.1"
2. Use debounce and distinct for ignoring frequent key input and duplicate input
3. Dispose previous API call for getting only latest search result

    override fun onCreateOptionsMenu(menu: Menu, inflater: MenuInflater) {
        inflater.inflate(R.menu.toolbar_menu, menu)

        // Associate searchable configuration with the SearchView
        val searchManager = requireContext().getSystemService(Context.SEARCH_SERVICE) as SearchManager
        searchView = menu.findItem(R.id.action_search).actionView as SearchView
        searchView.setSearchableInfo(
            searchManager.getSearchableInfo(requireActivity().componentName)
        )
        searchView.maxWidth = Integer.MAX_VALUE

        // listening to search query text change
        disposable = RxSearchView.queryTextChangeEvents(searchView)
            .debounce(750, TimeUnit.MILLISECONDS)
            .distinctUntilChanged()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                callApi(it.queryText().toString())
            }, {
                Timber.e(it)
            })

        super.onCreateOptionsMenu(menu, inflater)
    }

    private fun callApi(query: String){
        if(!apiDisposable.isDisposed){
            apiDisposable.dispose()
        }
        apiDisposable = mNetworkService.getAPI(query)
    }