我怎样才能有一个不会改变流的副作用订阅？

Question

设想：

我有一个数据列表（企业的交易）。如果满足以下两个条件，则此数据将被视为过时：

• 用户 (UI/View) 订阅了数据源 (an rx.Observable<List<Transaction>>)
• >= 自上次网络同步以来已过去 5 分钟。该值被序列化并从数据库中观察到。

注意：如果 UI/view 没有订阅数据源，并且 >= 5 五分钟过去了，我不想同步新数据（因为没有人在听）

以下代码示例的一些详细信息：

• data.observe() 返回 rx.Observable<List<Transaction>>
• void syncIfLast5Minutes() 将强制检查过去五分钟内是否有同步 - 如果没有，则将执行新的网络请求以获取新数据

现在，我可以轻松地执行以下操作：

data.observe()
    .doOnSubscribe(transactions -> syncIfLast5Minutes()

但这只会检查 UI 初始订阅时的数据是否过时。如果 UI 在 5 分钟后仍被订阅，则不会在此处触发自动刷新（但我希望它这样做）。

我正在寻找一个副作用运算符，它将

• 不影响原始流
• 订阅启动一个计时器，该计时器将监控数据何时过时（>= 5 分钟已过）
• 退订会取消此计时器

在 RxJava 中是否有一种惯用的方式来被动地实现这一点？

Answer 1

Observable.interval您可以通过将流与执行副作用（数据刷新）并已ignoreElements应用于它的流合并来实现您想要的目的。

这是一个工作示例：

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.subjects.PublishSubject;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        PublishSubject<String> data = PublishSubject.create();
        //refresh data every second for this demo
        //for your use case it's every 5 minutes
        data.mergeWith(Observable.interval(1, TimeUnit.SECONDS) 
                // cause timed side-effect that updates data stream
                .doOnNext(n -> data.onNext(n + "")) 
                .ignoreElements()
                .cast(String.class)) 
            .doOnNext(System.out::println) 
            .toBlocking() 
            .subscribe();
    }

}

如果多个订阅者正在使用该流，您可能想要查看，.share因此每 5 分钟仅发生一次刷新操作。