在即时搜索期间处理 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);