java - 处理RxJava即时搜索时中断异常

Question

我正在使用 RxJava 实现即时搜索。

问题:用户开始在搜索 EditText 上输入一些文本。用户暂停了一会儿。它通过了debounce操作条件。请求过程开始，但在获取最后一个请求的结果之前，用户开始输入更多字符或更改最后一个文本，现在将引发 java.io.InterruptedIOException 。 Observer 中的 onError 将被调用，搜索 EditText Observable 也将被释放，因此更改当前文本不再有任何效果，并且不会发生任何事情，因为 observable 已被释放。

预期行为:取消之前的请求并继续使用用户的新输入进行搜索。

代码:

        RxTextView.textChanges(etSearch)
                .skip(1)
                .map(CharSequence::toString)
                .doOnNext(this::handleSearchViewsVisibilityOnTextChange)
                .filter(string -> string.length() >= MIN_CHAR_TO_SEARCH)
                .debounce(SEARCH_QUERY_DELAY, TimeUnit.MILLISECONDS)
                .switchMap(this::search)
                .distinctUntilChanged()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(searchObserver);

    class SearchObserver implements Observer<Result> {

        @Override
        public void onNext(Result result) {

            if (result.getItemsCount() > 0) {
                resultList = result.getItemsList();
            } else {
                // No result
                resultList = new ArrayList<>();
            }
            searchAdapter.updateData(resultList);
        }

        @Override
        public void onError(Throwable throwable) {
            throwable.printStackTrace();
            List<Result> emptyData = Collections.emptyList();
            if (!isFinishing()) {
                searchAdapter.updateData(emptyData);
            }
        }
    }

Answer 1

您可以使用retryWhen像这样的东西:

      RxTextView.textChanges(etSearch)
                .skip(1)
                .map(CharSequence::toString)
                .doOnNext(this::handleSearchViewsVisibilityOnTextChange)
                .filter(string -> string.length() >= MIN_CHAR_TO_SEARCH)
                .debounce(SEARCH_QUERY_DELAY, TimeUnit.MILLISECONDS)
                .switchMap(this::search)
                .distinctUntilChanged()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .retryWhen((errors) -> {
                    errors.flatMap((error) -> {
                       error instanceof java.io.InterruptedIOException ? Observable.empty() : Observable.error(error)
                    }
                 })
                .subscribe(searchObserver);