Android Volley RequestFuture + RxJava2 Observable 重复

Question

在 MainActivity 中，我想每 5 秒刷新一次工具栏。
所以我用 Timer 和 Volley 异步请求实现了它，它就像一个魅力。

现在我更喜欢用 RxJava2 替换 Timer。
因此我做了这些步骤：

1. 我将 Async Volley 请求更改为 RequestFuture
2. 在 View 中，我使用 repeat() 方法将 Timer 更改为 Observable

但我的问题是： RX 每次重复发出的数据都与第一次相同。 为什么？

public Observable<String> doLiveRefresh2() {
    RequestFuture<String> future = RequestFuture.newFuture();
    VolleyStringRequest request = new VolleyStringRequest(Request.Method.POST
            , REFRESH_URL
            , future
            , future) {
        @Override
        protected Map<String, String> getParams() throws AuthFailureError {
            return getRefreshParams();
        }
    };

    VolleySingleton.getInstance().addToRequestQueue(request);
    return Observable.fromFuture(future, Constants.VOLLEY_SECONDS_TIME_OUT, TimeUnit.SECONDS, Schedulers.io());
}

---

final CompositeDisposable disposables = new CompositeDisposable();
Observable<LiveResponse> observable = new LiveRemoteModel().doLiveRefresh2()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .repeatWhen(completed -> completed.delay(Constants.TOOLBAR_BADGES_REFRESH_DELAY, TimeUnit.SECONDS))
                .map(LiveResponse::new)
                ;

        DisposableObserver<LiveResponse> disposableObserver =
                new DisposableObserver<LiveResponse>() {
                    @Override
                    public void onNext(@NonNull LiveResponse liveResponse) {
                        setToolbarBadges(liveResponse.getToolbarBadges());
                    }

                    public void onError(@NonNull Throwable e) {
                        Log.e("RX", "onError: ", e);
                    }

                    @Override
                    public void onComplete() {
                        Log.d("RX", "onComplete: ");
                    }
                };

        Disposable disposable = observable.subscribeWith(disposableObserver);
        disposables.add(disposable);

Answer 1

运营商repeatWhen 将重复值，并且不会重新订阅。为此，您需要一个稍微复杂的构造：

Observable<LiveResponse> observable = Observable
            .interval(Constants.TOOLBAR_BADGES_REFRESH_DELAY, TimeUnit.SECONDS))
            .subscribeOn(Schedulers.io())
            .flatMap(dummy -> new LiveRemoteModel().doLiveRefresh2())
            .map(LiveResponse::new)
            .observeOn(AndroidSchedulers.mainThread())
            ;