使用 RXJava2 / RXAndroid 2 和 Retrofit 进行轮询

Question

我想实现一个轮询服务，它每隔 nDelay 秒调用一次 REST Api，并在数据发生更改时通知所有订阅者。现在我的代码有点问题，因为它总是向我的消费者返回一个值，即使数据没有改变。

private Observable<List<HueLight>> pollingLightsObservable = null;

public Observable<List<HueLight>> getPollingLightsObservable() {
        if (pollingLightsObservable == null) {
            pollingLightsObservable = Observable.fromCallable(
                    () -> LightManager
                            .getInstance(context)
                            .getLights()
                            .blockingSingle())
                    //      .distinctUntilChanged( (l1, l1) -> !l1.equals(l2) )
                            .repeatWhen(o -> o.concatMap(v -> Observable.timer(1, TimeUnit.SECONDS)));

        }
        return pollingLightsObservable;
 }

启用或使用 distinctUntilChanged 不会改变任何东西。我把它放在我的repeatWhen之前还是之后都没关系。

由于我的 RetroFit 调用返回一个 Observable，我必须使用blockingSingle()。直接使用 Observable 会导致返回“4, 8, 12, 16, ..” 这个示例的项目：

LightManager.getInstance(context).getLights()
                            .repeatWhen(o -> o.concatMap(v -> Observable.timer(1, TimeUnit.SECONDS)))

目前我订阅了不同的课程/活动

   this.lightChangeSubscriber = PollingManager
                .getInstance(getContext())
                .getPollingLightsObservable()
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(hueLights -> {
                    {
                        Log.d(TAG, "Lights received successfully! Size=" + hueLights.size());
                    }
                });

我希望避免使用接口和计时器来创建轮询。你会推荐什么？

Answer 1

使用一些自定义过滤器怎么样？

public class FilterDuplicateHueConfig implements Predicate<HueConfig> {

   private HueConfig lastVal;
   @Override 
   public boolean test(HueConfig newVal) {
      if(lastVal == null) {
         lastVal = newVal;
         return true;
      }
      ... compare here the two values and return true/false appropriately...
   }
}