java - RxJava轮询+手动刷新

Question

我有一个列表想要每分钟刷新一次。 例如这里的用户列表:https://github.com/android10/Android-CleanArchitecture/blob/master/domain/src/main/java/com/fernandocejas/android10/sample/domain/interactor/GetUserList.java

我使用repeatWhen添加定期刷新:

  public Observable<List<User>> buildUseCaseObservable(Void unused) {
    return this.userRepository
        .users()
        .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
          @Override
          public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
            return objectObservable.delay(1, TimeUnit.MINUTES);
          }
        });
  }

这种方式工作得很好，每分钟调用一次 onNext。 但如果我想立即刷新此列表(由于用户的操作或通知)，我不知道如何执行该操作。

我应该取消/处置可观察对象并重新启动一个新的吗？ 谢谢

Answer 1

从您的代码中我了解到用户列表是在订阅时生成并发出的。

以下是我能想到的一些解决方案，而不是在您想要立即使用react的事件上取消订阅并重新订阅:

1. 不要使用 repeatWhen 运算符，而是使用 interval creation operator与 flatMap 结合，每分钟调用一个新 Observable 的订阅，并使用 merge 运算符添加对您感兴趣的其他事件的 react 。像这样的事情:

   @Test
   public void intervalObservableAndImmediateReaction() throws InterruptedException {
       Observable<String> obs = Observable.interval(1, TimeUnit.SECONDS)
                   .cast(Object.class)                                          
                   .mergeWith(
                             Observable.just("mockedUserClick")
                                       .delay(500, TimeUnit.MILLISECONDS))
                   .flatMap(
                            timeOrClick -> Observable.just("Generated upon subscription")
                            );
   
       obs.subscribe(System.out::println);
       Thread.currentThread().sleep(3000); //to see the prints before ending the test
   }
   

   或根据您的需要进行调整(但原理是相同的):

   Observable.interval(1, TimeUnit.MINUTES)
               .mergeWith(RxView.clicks(buttonView))
               .flatMap(timeOrClick -> this.userRepository.users());
   

2. 您可以像以前一样使用 flatMap 运算符，即使同时让您继续工作当前实现并且不合并到某个时间间隔 - 只需将您的工作代码保留在程序链的另一个区域中即可到RxBinding您选择的:

   RxView.touches(yourViewVariable)
         .flatMatp(motionEvent -> this.userRepository.users())
         .subscribe(theObserver);
   

   请注意，在此解决方案中，订阅是独立于两个可观察量完成的。如果你使用不同的观察者，或者管理一个主题或相关的事情，你的情况可能会更好。我运行的一个小测试显示，一个订阅者毫无问题地处理了 2 个不同的可观察量的订阅(在 Rxjava1 中 - 尚未检查 Rxjava2)，但对我来说感觉不太确定。