Android RXJava2改版处理分页

Question

public Observable<WatchList> syncWatchlistSection(String accountId) {
    String sessionToken = mLocal.getSessionId();
    int page = 1;

    return getWatchlistFromServerHelper(sessionToken, accountId, page).concatMap(new Function<WatchList, ObservableSource<? extends WatchList>>() {
        @Override
        public ObservableSource<? extends WatchList> apply(WatchList watchList) throws Exception {
            return Observable.fromArray(watchList);
        }
    });

}

private Observable<WatchList> getWatchlistFromServerHelper(String sessionToken, String accountId, int page) {
        return mRestAPI.fetchWatchlistOnServer(sessionToken, accountId, page).concatMap(new Function<WatchList, Observable<WatchList>>() {
            @Override
            public Observable<WatchList> apply(WatchList watchList) throws Exception {
                if (watchList.getPage() == watchList.getTotalPages()) {
                    return Observable.just(watchList);
                }
                return Observable.just(watchList).concatWith(getWatchlistFromServerHelper(sessionToken, accountId, page + 1));
            }
        });
    }

mSectionLogic.syncWatchlistSection(String.valueOf(account.getId()))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Observer<WatchList>() {
              ... so on ...

{
  page: 1,
  totalPages: 5,
  results: []
}

我做错了什么？我如何将结果与可观察的结果结合起来？最好不是用Lamda符号。

我查看的其他资源：-android rxjava2/retrofit2链接调用与分页令牌

Answer 1

你需要建立可观察的。您可以使用递归，但是如果您有大量的页，您的调用堆栈可能会溢出。在这里看到答案

类似于

private Observable<WatchList> getWatchlistFromServerHelper(String sessionToken, String accountId, int startPage) {
    int index = startPage;
    BehaviorSubject<Integer> pagecontrol = BehaviorSubject.create(startPage);
    Observable<WatchList> ret = pageControl.asObservable()
                                           .concatMap(new Function<Integer,Observable<WatchList>>()
                                                     {
                                                       Observable<WatchList> apply(Integer page) {
                                                            mRestAPI.fetchWatchlistOnServer(sessionToken, accountId, page)
                                                                    .doOnNext(new Function<watchList -> 
                                                                            (watchList.getPage() == watchList.getTotalPages() ? 
                                                                             Observable.<WatchList>empty().doOnCompleted(()->pageControl.onCompleted()) :
                                                                             pageControl.onNext(page+1)));
                                                        }
                                                     }
                                                     );
    return ret;
}

应该管用。