Android：使用repeatWhen、takeUntil和带改造的过滤器通过rx-java无限滚动

我正在使用 Retrofit 2.2 和 RxJava。 分页是这样工作的：我得到第一批数据，我必须请求具有相同参数的第二批数据，除了最后更新日期，然后如果我得到空或同一批数据，则意味着有没有更多的物品。我发现了这篇很棒的文章https://medium.com/@v.danylo/server-polling-and-retrying-failed-operations-with-retrofit-and-rxjava-8bcc7e641a5a#.40aeibaja https://medium.com/@v.danylo/server-polling-and-retrying-failed-operations-with-retrofit-and-rxjava-8bcc7e641a5a#.40aeibaja关于如何去做。所以我的代码是：

private Observable<Integer> syncDataPoints(final String baseUrl, final String apiKey,
        final long surveyGroupId) {
    final List<ApiDataPoint> lastBatch = new ArrayList<>();
    Timber.d("start syncDataPoints");
    return loadAndSave(baseUrl, apiKey, surveyGroupId, lastBatch)
            .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
                @Override
                public Observable<?> call(final Observable<? extends Void> observable) {
                    Timber.d("Calling repeatWhen");
                    return observable.delay(5, TimeUnit.SECONDS);
                }
            })
            .takeUntil(new Func1<List<ApiDataPoint>, Boolean>() {
                @Override
                public Boolean call(List<ApiDataPoint> apiDataPoints) {
                    boolean done = apiDataPoints.isEmpty();
                    if (done) {
                        Timber.d("takeUntil : finished");
                    } else {
                        Timber.d("takeUntil : will query again");
                    }
                    return done;
                }
            })
            .filter(new Func1<List<ApiDataPoint>, Boolean>() {
                @Override
                public Boolean call(List<ApiDataPoint> apiDataPoints) {
                    boolean unfiltered = apiDataPoints.isEmpty();
                    if (unfiltered) {
                        Timber.d("filtered");
                    } else {
                        Timber.d("not filtered");
                    }
                    return unfiltered;
                }
            }).map(new Func1<List<ApiDataPoint>, Integer>() {
                @Override
                public Integer call(List<ApiDataPoint> apiDataPoints) {
                    Timber.d("Finished polling server");
                    return 0;
                }
            });
}

private Observable<List<ApiDataPoint>> loadAndSave(final String baseUrl, final String apiKey,
        final long surveyGroupId, final List<ApiDataPoint> lastBatch) {
    return loadNewDataPoints(baseUrl, apiKey, surveyGroupId)
            .concatMap(new Func1<ApiLocaleResult, Observable<List<ApiDataPoint>>>() {
                @Override
                public Observable<List<ApiDataPoint>> call(ApiLocaleResult apiLocaleResult) {
                    return saveToDataBase(apiLocaleResult, lastBatch);
                }
            });
}


private Observable<ApiLocaleResult> loadNewDataPoints(final String baseUrl, final String apiKey,
        final long surveyGroupId) {
    Timber.d("loadNewDataPoints");

    return Observable.just(true).concatMap(new Func1<Object, Observable<ApiLocaleResult>>() {
        @Override
        public Observable<ApiLocaleResult> call(Object o) {
            Timber.d("loadNewDataPoints call");
            return restApi
                    .loadNewDataPoints(baseUrl, apiKey, surveyGroupId,
                            getSyncedTime(surveyGroupId));
        }
    });
}

正如你所看到的，有趣的方法是loadNewDataPoints，我希望它被调用，直到没有更多的数据点。如你看到的Observable.just(true).concatMap是一个黑客，因为如果我删除这个连续映射restApi.loadNewDataPoints(....)虽然在日志中我可以看到 api 确实被调用，但没有被调用，但使用相同的旧参数，当然它返回与第一次相同的结果，因此同步停止，saveToDataBase 确实被调用得很好。通过我的黑客，它可以工作，但我想了解为什么它不能以其他方式工作，以及是否有更好的方法来做到这一点。多谢！

因此，我编写了此类 API（称为 Keyset Pagination）并针对它们实现了 Rx 客户端。

这是BehaviorSubjects 有用的情况之一：

S initialState = null;
BehaviorProcessor<T> subject = BehaviorProcessor.createDefault(initialState);
return subject
  .flatMap(state -> getNextElements(state).singleOrError().toFlowable(), Pair::of, 1)
  .serialize()
  .flatMap(stateValuePair -> {
      S state = stateValuePair.getLeft();
      R retrievedValue = stateValuePair.getRight();
      if(isEmpty(retrievedValue)) {
         subject.onComplete();
         return Flowable.empty();
      } else {
         subject.onNext(getNextState(state, retrievedValue));
         return Flowable.just(retrievedValue);
      }
    }
   .doOnUnsubscribe(subject::onCompleted)
   .map(value -> ...)

Here

• getNextElement基于状态执行网络调用并返回具有单个值的反应流
• isEmpty判断返回值是否为空，表示元素结束
• getNextState将传入的状态与检索到的值相结合来确定下一个状态getNextElement.

如果发生错误（它将被传播）并且您在结束之前取消订阅（查询将被终止），它将正常工作。

当然，在您的具体情况下，这些不需要是单独的方法或复杂的类型。