rxjava interval() 在某些事件后重置时间

Question

我正在用 RxJava 制作一个 Android 应用程序，在其中一个页面中，我有一个按钮，当按下时，页面会刷新。如果用户在此期间没有按下按钮，我还希望每 10 秒自动刷新一次。但是当用户单击按钮时，我希望自动刷新操作在单击后 10 秒后发生。而不是继续其自己的 10 秒间隔。例如，在第 0 秒，应用程序进行自动刷新，然后在第 3 秒，用户按下按钮。然后自动刷新应该发生在第 13 秒、第 23 秒等。我知道有一个 interval() 运算符会以一定的时间间隔发出项目。不过好像没办法"reset"开始时间。它有点像取消订阅并再次订阅 interval() Observable。一段代码就像

Observable<Long> intervalObservable = Observable.inteval(10, TimeUnit.SECONDS)
RxView.click(refreshButton).map(ignored -> 0L)
      .merge(intervalObservable)
      .subscibe(ignore -> performRefresh());

如果有办法 "unmerge" 和 intervalObservable，那么我可以在 onNext 中取消合并，然后再次合并。但是好像没有。我怎样才能做到这一点？

Answer 1

最简单的解决方案不是每次单击按钮时都取消订阅旧的 Observable 并再次订阅吗？

   Subscription subscription= null;  
   Observable<Long> interval = nul;


 public void subscribe(){

    interval = Observable
            .interval(10000L, TimeUnit.MILLISECONDS);
    subscription = interval.subscribe(item -> System.out.println("doing something")); 

}

public void eventClick(){
    if(clickButtin){
        subscription.unsubscribe();
        subscribe();
    }
}

Answer 2

您可以使用 switchMap 运算符很好地完成此操作。每次按下按钮时，它都会切换到间隔可观察对象的新订阅 - 这意味着它将重新开始。之前的订阅会自动删除，因此不会有多个间隔运行.

Observable<Long> intervalObservable = Observable.interval(10, TimeUnit.SECONDS);

RxView.clicks(refreshButton)
    .switchMap(ignored -> {
        return intervalObservable
            .startWith(0L)                 // For an immediate refresh
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext(x -> performRefresh());
    })      
    .subscribe();

startWith 向间隔添加一个立即值（单击按钮时立即刷新），observeOn 确保刷新发生在主线程上（这很重要，因为interval 将在后台线程上发出）。

Update：vman注意到这个实现只有在用户第一次点击按钮后才开始刷新。以下选项将在订阅后立即刷新，然后每 10 秒刷新一次，直到单击按钮 – 此时它将再次立即刷新，然后继续每 10 秒更新一次。

Observable<Long> intervalObservable = Observable.interval(10, TimeUnit.SECONDS)
    .startWith(0L)  // For an immediate refresh
    .observeOn(AndroidSchedulers.mainThread())
    .doOnNext(x -> performRefresh());

Observable<Long> buttonClickedObservable = RxView.clicks(refreshButton)
    .map(e -> 0L)  // To make the compiler happy
    .switchMap(ignored -> Observable.error(new RuntimeException("button pressed")));

Observable.merge(intervalObservable, buttonClickedObservable)
    .retry()
    .subscribe();

在这个实现中，一个进行刷新的 Observable 被立即订阅（导致立即刷新，并且每 10 秒重复一次），它与一个 Observable 合并，只要点击刷新按钮就会出错。最后的 retry 将导致整个事件在错误（单击按钮）时重新订阅 - 导致一切重新开始。

rxjava interval() 在某些事件后重置时间

rxjava interval() reset time after some event

java

android

rx-java

rx-android