RxJava 轮询 + 手动刷新答案

【问题标题】：RxJava polling + manual refreshRxJava 轮询 + 手动刷新
【发布时间】：2017-09-20 09:53:50
【问题描述】：

我有一个想要每分钟刷新一次的列表。例如这里的用户列表：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);
          }
        });
  }

这种方式运行良好，每分钟调用一次。但是如果我想立即刷新这个列表（由于用户的操作或由于通知），我不知道如何执行。

我应该取消/处置 observable 并重新启动一个新的吗？谢谢

【问题讨论】：

标签： java android rx-java polling

【解决方案1】：

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

这里有一些我能想到的解决方案，而不是取消订阅并重新订阅您想要立即做出反应的事件：

不要使用 repeatWhen 运算符，而是使用 interval creation operator 与 flatMap 组合来每分钟调用一个新的 Observable 订阅，并使用 merge 运算符添加对您在其中的其他事件的反应感兴趣。像这样的：

@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());

您可以像以前一样使用 flatMap 运算符，即使在保持当前实现工作且不合并到间隔的情况下 - 只需保留您的工作代码并在程序的另一个区域将其链接到您选择的 RxBinding ：
```
RxView.touches(yourViewVariable)
      .flatMatp(motionEvent -> this.userRepository.users())
      .subscribe(theObserver);
```
请注意，在此解决方案中，订阅是独立于两个 observable 完成的。如果你使用不同的观察者，或者管理一个主题或在那条线上的东西，你可能会更好。我运行的一个小测试显示，一个订阅者可以毫无问题地处理订阅 2 个不同的 observables（在 Rxjava1 中 - 还没有在 Rxjava2 中签入），但对我来说感觉不太好。

【讨论】：

使用interval 和mergeWith 似乎很有趣，谢谢。如何手动触发“刷新”（而不是使用 RxView）？
您可以使用一个主题，然后在程序中的任何位置调用它的onNext。

【解决方案2】：

如果您不关心在其他可观察对象之一发出数据后调整刷新时间，您可以执行以下操作：

    // Specific example of a user manually requesting
    val request = Observable.create<String> { emitter ->
        refresh.setOnClickListener {
            emitter.onNext("Click Request")
        }
    }
            .observeOn(Schedulers.io())
            .flatMap {
                userRepository.users()
            }

    // Refresh based off of your original work, could use something like interval as well
    val interval = userRepository.users()
            .subscribeOn(Schedulers.io())
            .repeatWhen { objectObservable ->
                objectObservable.delay(1, TimeUnit.MINUTES)
            }

    // Combine them so that both emissions are received you can even add on another source
    Observable.merge(request,interval)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                contents.text = it.toString()
            }, {
                contents.text = it.toString()
            },{
                println(contents.text)
            })

那么你就不用每次都去dispose和resubscribe了

【讨论】：

如何将 Observable 与 Observable> 合并？