RxJava + 改造 + 轮询

Question

我有一个改造电话，想每 30 秒调用一次。为此，我使用Observable.interval(0, 30, TimeUnit.SECONDS)

Observable
    .interval(0, 30, TimeUnit.SECONDS)
    .flatMap(x -> RestApi.instance().getUsers())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(list -> {
                    // ...
               },
               error -> Timber.e(error, "can't load users"));

我的问题：如果 api 调用失败，onError 被调用并且订阅取消订阅并且轮询不再起作用:-(

为了捕捉 api 错误，我添加了 retryWhen

Observable
    .interval(0, 30, TimeUnit.SECONDS)
    .flatMap(x -> RestApi.instance().getUsers()
                         .retryWhen(errors -> errors
                             .flatMap(error -> Observable.timer(15, TimeUnit.SECONDS))))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(list -> {
                   // ...
               },
               error -> Timber.e(error, "can't load users"));

这会捕获错误，但随着时间的推移，我会收到多个 api 调用。每 30 秒我收到一个新的轮询信号，该信号以新的 api 请求结束。但如果 api 请求失败，它会自行重试。所以我有一个新请求以及所有重试。

我的问题：如何在不取消订阅 poll 信号的情况下处理 api 错误？

Answer 1

您可以使用此代码，在此代码中实现请求之间的次数尝试和时间延迟

private static int COUNTER_START = 1;
private static final int ATTEMPTS = 6;
private static final int ORIGINAL_DELAY_IN_SECONDS = 2;

remoteData.getAllRides(idSearch)
            .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
                        @Override
                        public Observable<?> call(Observable<? extends Void> observable) {
                            return observable.flatMap(new Func1<Void, Observable<?>>() {
                                @Override
                                public Observable<?> call(Void aVoid) {
                                    if(COUNTER_START > ATTEMPTS){
                                        throw new RuntimeException();
                                    }
                                    COUNTER_START++;
                                    return Observable.timer(ORIGINAL_DELAY_IN_SECONDS, TimeUnit.SECONDS);
                                }
                            });
                        }
                    })
            .takeUntil(new Func1<RideResponse, Boolean>() {
                @Override
                public Boolean call(RideResponse rideResponse) {
                    return rideResponse.getState().equals("finished");//this is the validation finish polling

                }
            }).filter(new Func1<RideResponse, Boolean>() {
                @Override
                public Boolean call(RideResponse rideResponse) {
                    return rideResponse.getState().equals("finished"); //this is the validation finish polling
                }
            }).map(rideResponse -> Log.e("",rideResponse.toString()))
        .doOnError(err -> Log.e("Polling", "Error retrieving messages: " + err));

Answer 2

阅读如何正确使用retryWhen 和repeatWhen。 http://blog.danlew.net/2016/01/25/rxjavas-repeatwhen-and-retrywhen-explained/

以及如何使用onError 运算符： http://blog.danlew.net/2015/12/08/error-handling-in-rxjava/

使用 Rx 真的很容易 :) 我不会给你一个最终的解决方案，只是玩弄它并尝试理解这里的流程。

Answer 3

如果您希望您的getUsers 请求在请求失败时不以onError 结尾，而不是返回Observable<yourUserType> getUsers()，请使其返回Observable<Response<yourUserType>> getUsers()。这样就可以拦截Response对象中的网络错误。

此方法仅在您使用改造 2.x 时有效

Answer 4

您可以使用onErrorResumeNext 或 onExceptionResumeNext 并传递“错误”值。 你可以根据自己的需要寻找其他错误处理here