使用 RxAndroid 的智能轮询

Question

我想要实现的目标：我想通过每 5 分钟轮询一次从网络上轮询一些资源，但前提是订阅了观察者。我使用行为主体和可观察间隔进行池化。我设法实现了它，但我是 Rx 的新手，我认为它可以做得更好。

这就是我的做法：

private BehaviorSubject<String> observable;
private Subscription> subscription;
public Subscription subscribe(final Action1<String> action) {
    if (observable == null) {
        observable = BehaviorSubject.create();
    }
    if (subscription == null) {
        Observable<String> source = Observable.interval(5, TimeUnit.MINUTES).map(new Func1<Long, String>() {
            @Override
            public String call(Long aLong) {
                return getDataFromServer();
            }
        }).observeOn(AndroidSchedulers.mainThread());
        subscription = source.subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                if (observable.hasObservers()) {
                    observable.onNext(s);
                } else {
                    subscription.unsubscribe();
                    subscription = null;
                }
            }
        });
    }
    return observable.subscribe(action);
}

这个想法：- 我有一个可观察的轮询源和另一个可观察的源，客户端可以订阅(使用行为主体实现 - 所以他们总是得到最新的数据(。当源可观察量发出一些东西时，如果行为主体有订阅者，它就会被传递，否则什么都不会传递，我取消订阅源，所以它会停止。

Answer 1

怎么样：

Observable<String> observable = Observable.interval(0, 5, TimeUnit.SECONDS)
    .doOnNext(new LoggingAction1<Long>("doOnNext"))
    .flatMap(new Func1<Long, Observable<String>>() {
        @Override
        public Observable<String> call(Long pulse) {
            return Observable.just(String.format("Request %d", pulse));
        }
    })
    .replay(1)
    .refCount();

我认为它可以完成您希望您的设置执行的所有操作：

• 只要没有Subscribers它就什么都不做。
• 当第一个Subscriber订阅时，interval将启动并立即发出一个值，然后每 5 秒发出一个值。
• 新Subscriber将立即获得最后一项，然后获得所有后续项。
• 无论有多少Subscribers，只会启动一个interval - 因此每 5 秒只会执行一个网络请求。
• 当所有Subscribers都取消订阅后，interval将停止发送项目。