使用RxAndroid进行智能轮询

Question

我想要实现的目标:我希望每5分钟轮询一次来自网络的资源,但只有在有观察者订阅的情况下.我使用BehaviorSubject和interval observable进行池化.我设法实现了它,但我是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);
}

这个想法: - 我有一个可观察的轮询源和客户可以订阅的另一个observable(使用BehaviourSubject实现 - 所以他们总是得到最新的数据).当source observable发出一些东西时,如果行为主体有订阅者,则传递给它,否则什么也没有传递,我取消订阅源,所以它会停止.

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将立即获得最后一个项目,然后是所有后续项目.
• 只会interval启动一个- 因此每5秒钟只会执行一次网络请求 - 无论有多少网络请求Subscribers.
• 当所有人Subscribers都取消订阅时,interval将停止发射物品.