使用 RxJava 轮询服务器 - 在获得服务器响应后重复答案

【问题标题】：Server polling using RxJava - repeat after geting server response使用 RxJava 轮询服务器 - 在获得服务器响应后重复
【发布时间】：2016-10-18 03:17:34
【问题描述】：

目前我正在尝试使用 RxJava 实现服务器轮询，我已经对如何在收到服务器响应后重复整个链进行了一些研究，我尝试使用 repeat()，它可以工作但不能完美地工作，原因是它使api调用了很多次，服务器在发送到客户端之前需要额外的时间来处理数据，但是我们不知道确切的时间，所以我们不能使用repeatWhen()来给出具体的时间。我唯一能做的就是等待 api 响应。

任何建议将不胜感激！

以下为sn-p代码：

retrofitService.requestPolling()
              .repeat()   // do not wait to call server so many times
              .takeUntil(new Func1<PollResponse, Boolean>() {
                      @Override
                      public Boolean call(PollResponse pollResponse) {
                            return pollResponse.mComplete;
                       }
               })
             .doOnNext(new Action1<FlightSearchPollResponse>() {
                @Override
                public void call(pollResponse pollResponse) {
                      // update UI here
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<PollResponse>() {
                @Override
                public void onCompleted() {

                    }
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(PollResponse pollResponse) {

            } );

编辑：我是 RxJava 的新手，刚刚得到了一个名为 BackPressure 的主题，并且有很多文章解释了如何处理它，因为我不想缓存这个响应，似乎主题是一个不错的选择，它允许你控制什么时候拉。

http://akarnokd.blogspot.com/2015/06/subjects-part-1.html

感谢@Gary LO

【问题讨论】：

标签： rx-java reactive-programming polling

【解决方案1】：

应该有很多方法。我想分享其中之一。

创建单独的信号流PublishSubject pollingSignal
将信号转换为 api 调用

发布信号再做一次。

final PublishSubject<Boolean> pollingSignal = PublishSubject.create();

final Observable<PollResponse> apiResponse = retrofitService.requestPolling();

pollingSignal
  .flatMap(x -> apiResponse)
  .subscribe(new Observer<PollResponse>() {
    @Override
    public void onCompleted() {}

    @Override
    public void onError(Throwable throwable) {}

    @Override
    public void onNext(PollResponse integer) {
      // start the next polling 
      pollingSignal.onNext(true);
  }
});

// start the first polling 
pollingSignal.onNext(true);

玩得开心！

备注使用PublishSubject<Boolean> 而不是PublishSubject<Void> 是因为我觉得使用pollingSignal.onNext(null) 不舒服。

但在 Kotlin 中，我可以使用 PublishSubject<Unit> 和 pollingSignal.onNext(Unit)

【讨论】：