RxJava 分页

Question

我对 rxJava 还很陌生。在我的 api 响应中，我获得了有关总页数和当前页码的信息，例如：

 "pages": 22,
 "page": 1,

我正在使用 Retrofit 在数据层做 api 调用，我的 api 服务是这样的：

@GET("/story")
    Observable <StoryCollectionEntity> storyCollection(
              @Query("feed_ids") String feed_items, 
              @Query("page") int  page);

然后：

   public Observable<StoryCollectionEntity> storyCollection() {
        return mUserApi.storyCollection(items,page);
}

我在域层做了这样的订阅：

public void execute(Subscriber UseCaseSubscriber) {
        this.subscription = this.buildUseCaseObservable()
                 .subscribeOn(Schedulers.from(threadExecutor))
                .observeOn(postExecutionThread.getScheduler())
                .subscribe(UseCaseSubscriber);

    }
 @Override public Observable buildUseCaseObservable() {
    return this.userRepository.stories();
  }

我正在弄清楚如何通过发出下一页的结果来让这个观察者对 recyclerView 滚动事件做出反应。即第 2 页到第二次滚动事件和第 3 页在第 3 次滚动...等

Answer 1

我将粘贴在我正在使用的解决方案下方 - 该解决方案也使用 Retrofit。 在下面的示例中，我使用 Retrofit 从 GitHub API 创建 所有 个公共存储库的 Observable。

我相信这里的 技巧 是我在 Observable 上递归并且与另一个 Observable 连接本质上是一个触发器。

这个触发器可以通过多种方式发出（可能是当用户滚动到页面底部时），但是在这个例子中，一旦处理了整个页面，它就会立即发出。

/***
 * This is a helper wrapper Subscriber that helps you lazily defer
 * continuous paging of a result set from some API.
 * Through the use of a {@link Subject}, it helps notify the original {@link Observable}
 * when to perform an additional fetch.
 * The notification is sent when a certain count of items has been reached.
 * Generally this count represents the page.
 * @param <T> The event type
 */
@Data
public class PagingSubscriber<T> extends Subscriber<T> {

    private final Subject<Void,Void> nextPageTrigger = PublishSubject.create();
    private final long pageSize;
    private long count = 0;
    private final Subscriber<T> delegate;

    /***
     * Creates the {@link PagingSubscriber}
     * @param pageSize
     * @param delegate
     */
    public PagingSubscriber(long pageSize, Subscriber<T> delegate) {
        this.pageSize = pageSize;
        this.delegate = delegate;
    }

    public Observable<Void> getNextPageTrigger() {
        return nextPageTrigger;
    }

    @Override
    public void onStart() {
        delegate.onStart();
    }

    @Override
    public void onCompleted() {
        delegate.onCompleted();
    }

    @Override
    public void onError(Throwable e) {
        delegate.onError(e);
    }

    @Override
    public void onNext(T t) {
        count+=1;
        if (count == pageSize) {
            nextPageTrigger.onNext(null);
            count= 0;
        }
        delegate.onNext(t);
    }

}

@Data
public class GitHubRepositoryApplication {

    private final GitHubService gitHubService;

    @Inject
    public GitHubRepositoryApplication(GitHubService githubService) {
        this.gitHubService = githubService;
    }

    public Observable<GitHubRepository> printAllRepositories(Observable<Void> nextPageTrigger) {
        return printRepositoryPages(GitHubService.FIRST_PAGE, nextPageTrigger)
                .flatMapIterable(r -> r.body());
    }


    public Observable<Response<List<GitHubRepository>>> printRepositoryPages(String startingPage, Observable<Void> nextPageTrigger) {
        return gitHubService.listRepos(startingPage)
                .concatMap(response -> {
                    Optional<String> nextPage = Optional.ofNullable(response.headers().get(HttpHeaders.LINK))
                            .flatMap(header -> GitHubServiceUtils.getNextToken(header));

                    if (!nextPage.isPresent()) {
                        return Observable.just(response);
                    }
                    return Observable.just(response)
                            .concatWith(nextPageTrigger.limit(1).ignoreElements().cast(Response.class))
                            .concatWith(printRepositoryPages(nextPage.get(), nextPageTrigger));
                });
    }

    public static void main(String[] args) {
        Injector injector = Guice.createInjector(new GitHubModule());

        GitHubRepositoryApplication app = injector.getInstance(GitHubRepositoryApplication.class);

        Subscriber<GitHubRepository> subscriber = new Subscriber<GitHubRepository>() {

            private final Logger log = LoggerFactory.getLogger(getClass());

            @Override
            public void onStart() {
                log.debug("STARTING");
                request(1l);//we need to begin the request
            }

            @Override
            public void onCompleted() {
                log.debug("COMPLETED");
            }

            @Override
            public void onError(Throwable e) {
                log.error("ERROR",e);
            }

            @Override
            public void onNext(GitHubRepository gitHubRepository) {
                log.debug("{}",gitHubRepository);
                request(1l);//we need to make sure we have asked for another element
            }
        };

        PagingSubscriber<GitHubRepository> pagingSubscriber = new PagingSubscriber<>(GitHubService.PAGE_SIZE, subscriber);

        //In order for the JVM not to quit out, we make sure we turn our Observable to
        //a BlockingObservable, so that all of it will finish.
        Observable<GitHubRepository> observable =
                app.printAllRepositories(pagingSubscriber.getNextPageTrigger());
        observable.toBlocking().subscribe(pagingSubscriber);

    }

}