Repeat与Retry、repeatWhen与retryWhen

repeat()接收到onCompleted()事件后触发重订阅。
retry()接收到onError()事件后触发重订阅。

要实现一个延迟数秒的重订阅？或者想通过错误来决定是否重订阅呢？这种情况下就需要repeatWhen和retryWhen。

repeatWhen 实现变长时延轮询

可以通过repeatWhen来实现轮询，是因为它提供了重订阅的功能，而重订阅有两点要素：

上游告诉我们一次订阅已经完成，这就需要上游回调 onComplete函数。
我们告诉上游是否需要重订阅，通过repeatWhen的Function函数所返回的Observable确定，如果该Observable发送了onComplete或者onError则表示不需要重订阅，结束整个流程；否则触发重订阅的操作。

Returns an Observable that emits the same values as the source ObservableSource with the exception of an onComplete. An onComplete notification from the source will result in the emission of a void item to the ObservableSource provided as an argument to the notificationHandler function. If that ObservableSource calls onComplete or onError then repeatWhen will call onComplete or onError on the child subscription. Otherwise, this ObservableSource will resubscribe to the source ObservableSource.

image.png

repeatWhen的难点在于如何定义它的Function参数：

Function的输入是一个Observable<Object>，输出是一个泛型ObservableSource<?>。
如果输出的Observable发送了onComplete或者onError则表示不需要重订阅，结束整个流程；否则触发重订阅的操作。
对于每一次订阅的数据流Function 函数只会回调一次，并且是在onComplete的时候触发。
在Function函数中，必须对输入的 Observable<Object>进行处理，这里我们使用的是flatMap操作符接收上游的数据。

不需要重订阅时，有两种方式：

返回Observable.empty()，发送onComplete消息，但是DisposableObserver并不会回调onComplete。
返回Observable.error(new Throwable("Polling work finished"))，DisposableObserver的onError会被回调，并接受传过去的错误信息。

Observable<Long> observable = Observable.just(0L).doOnComplete(new Action() {

    @Override
    public void run() throws Exception {
        Log.i("Observable", "Observable");
    }

}).repeatWhen(new Function<Observable<Object>, ObservableSource<Long>>() {

    private long mRepeatCount;

    @Override
    public ObservableSource<Long> apply(Observable<Object> objectObservable) throws Exception {
        //必须作出反应，这里是通过flatMap操作符。
        return objectObservable.flatMap(new Function<Object, ObservableSource<Long>>() {

            @Override
            public ObservableSource<Long> apply(Object o) throws Exception {
                if (++mRepeatCount > 4) {
                    //return Observable.empty(); //发送onComplete消息，无法触发下游的onComplete回调。
                    return Observable.error(new Throwable("Polling work finished")); //发送onError消息，可以触发下游的onError回调。
                }
                Log.d(TAG, "startAdvancePolling apply");
                return Observable.timer(3000 + mRepeatCount * 1000, TimeUnit.MILLISECONDS);
            }

        });
    }

});

Returns an Observable that emits the same values as the source ObservableSource with the exception of an onError. An onError notification from the source will result in the emission of a Throwable item to the ObservableSource provided as an argument to the notificationHandler function. If that ObservableSource calls onComplete or onError then retry will call onComplete or onError on the child subscription. Otherwise, this ObservableSource will resubscribe to the source ObservableSource.

image.png

Observable<String> observable = Observable.defer(new Callable<ObservableSource<String>>() {
    @Override
    public ObservableSource<String> call() throws Exception {
        String cacheToken = TokenLoader.getInstance().getCacheToken();
        return Observable.just(cacheToken);
    }
}).flatMap(new Function<String, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(String token) throws Exception {
        return getUserObservable(token);
    }
}).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {

    private int mRetryCount = 0;

    @Override
    public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
        return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {

            @Override
            public ObservableSource<?> apply(Throwable throwable) throws Exception {
                Log.d(TAG, index + ":" + "发生错误=" + throwable + ",重试次数=" + mRetryCount);
                if (mRetryCount > 0) {
                    return Observable.error(new Throwable(ERROR_RETRY));
                } else if (ERROR_TOKEN.equals(throwable.getMessage())) {
                    mRetryCount++;
                    return TokenLoader.getInstance().getNetTokenLocked();
                } else {
                    return Observable.error(throwable);
                }
            }
        });
    }
});

defer：读取缓存中的token信息，这里调用了TokenLoader中读取缓存的接口，而这里使用defer操作符，是为了在重订阅时，重新创建一个新的Observable，以读取最新的缓存token信息。
retryWhen：使用重订阅的方式来处理token失效时的逻辑，这里分为三种情况：重试次数到达，那么放弃重订阅，直接返回错误；请求token接口，根据token请求的结果决定是否重订阅；其它情况直接放弃重订阅。