android - RxJava : how to recover from errors in flatMap operator

Question

我有一个 EditText，用户可以在其中输入搜索查询，我想在用户输入内容时在我的服务器上执行即时搜索。

我尝试使用 RxJava 来实现，如下所示:

RxTextView.textChanges(editQuery) // I'm using RxBinding for listening to text changes
    .flatMap(new Func1<CharSequence, Observable<UserPublic[]>>() {
        @Override
        public Observable<UserPublic[]> call(CharSequence query) {
            return api.searchUsers(query); // I'm using Retrofit 1.9 for network calls. searchUsers returns an Observable<UserPublic[]>
        }
    })
    .subscribe(Observers.create(
        new Action1<UserPublic[]>() {
            @Override
            public void call(UserPublic[] userPublics) {
                processResult(userPublics);
            }
        })
        , new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                processError(throwable);
            }
    });

问题是，如果网络调用遇到错误，整个可观察对象就会停止。因此，当用户继续输入时，什么也不会发生。

我如何修改这段代码以便:

1. 只要出现网络问题，就会调用processError
2. 但是当用户继续输入时，新的网络调用继续发出(导致 processResult/processError 再次)

Answer 1

使用 retryWhen()运算符并在订阅之前将其附加到可观察链。请注意 retryWhen() 的参数是一个接受 Observable<Throwable> 的函数并返回 Observable<?> .运算符使用 onNext() 时返回的类型无关紧要结果启动重试和 onError()或 onCompleted()结果终止链。

这是一个等待 5 秒并再次尝试的天真的应用程序:

observable
  .retryWhen( errorObservable -> errorObservable.delay( 5, TimeUnit.SECONDS ) )
  .subscribe();

这是一个不那么天真的操作，它在超时时重试，如果出现 IOException 则失败。发生:

observable
  .retryWhen( errorObservable -> errorObservable.flatMap( throwable -> { // (1)
       if ( throwable instanceof IOException ) {
         return Observable.error( throwable ); // (2)
       }
       return Observable.just(1); // (3)
     } )
  .subscribe();

1. 使用 flatMap()允许您推迟决定，直到您知道您正在处理的错误类型。
2. 返回的可观察对象抛出提供的错误或您希望更好地描述您的问题的其他内容。
3. 提供一个简单地执行 onNext() 的可观察对象告诉retryWhen()运算符(operator)重新订阅原始可观察对象。