我们如何使用 RxJava 进行分页 Web 服务调用，其中每个页面都依赖于前一个页面的响应，而无需递归？

Question

Web 服务 APIs 有时使用分页，其中 Web 服务调用的参数指示要检索的页面。这些大致可以分为两种：

请求页面的参数独立于任何给定的页面响应（例如，"give me page #3, with a page size of 10"）
请求页面的参数取决于先前页面响应的那些（例如，“给我标识符为 foo 的项目之后的下 10 个项目）

很好地涵盖了第一个场景，其中 Web 服务只需要一个页码，我们需要从任何给定页面的响应中确定我们是否完成。

涵盖了第二种情况，但它依赖于递归，因此对于大型数据集，我们将死于 WhosebugError.

中继兼容的 GraphQL 支持的 Web 服务（例如，GitHub 的 API）将大量使用第二种情况，因为 Relay's specification for pagination 要求您提供"cursor" 从先前的响应中获取该光标位置之后的下一个项目。所以，我试图为此找出一种非递归方法，它仍然将所有内容都包装到一个主控中 Observable，就像这两个答案所做的那样。

Answer 1

如果 Web 服务 API 正在阻塞或者您愿意阻塞，那么解决方案很简单

Observable.generate(() -> new ApiResponse(page), (s, emitter) -> {
    ApiResponse r = getResults(s.next);            
    emitter.onNext(r);
    if (r.next == null) emitter.onComplete();
    return r;
});

使用递归答案的符号。

如果不需要阻塞，您可以像这样使用 RxJava2Extensions 中的 FlowableTransformers.expand

Flowable
    .just(new ApiResponse(page))
    .compose(FlowableTransformers.expand(r -> r.next == null ? Flowable.empty() : getResults(r.next)));

Answer 2

我不知道我是否理解正确，但我相信你可以在 RxJava 上使用 SyncOnSubscribe.createStateful 来解决这个问题。

查看我的样本：

class SimpleTest {
  @Test
  fun testRequestTenPages() {
    getPaginatedDataFromApi()
        .take(10)
        .subscribe { println(it) }
  }

  fun apiCall(previous: Response? = null) : Response {
    return previous?.let {
      val newPage = it.page + 1
      previous.copy(id = "${it.id}_$newPage", page = newPage)
    } ?: Response("1", 1)
  }    

  fun getPaginatedDataFromApi(): Observable<Response> {
    val syncOnSubscribe = SyncOnSubscribe.createStateful<Response?, Response>(
        { null },
        { previous, observer ->
          val response = apiCall(previous)
          observer.onNext(response)
          return@createStateful response
        }
    )

    return Observable.create(syncOnSubscribe)
  }

  data class Response(val id: String, val page: Int)
}

我正在创建一个有状态的可观察对象，它将最后一个响应保存为状态，并使用它来生成下一个响应。

运行此测试您将看到以下输出：

Response(id=1, page=1)
Response(id=1_2, page=2)
Response(id=1_2_3, page=3)
Response(id=1_2_3_4, page=4)
Response(id=1_2_3_4_5, page=5)
Response(id=1_2_3_4_5_6, page=6)
Response(id=1_2_3_4_5_6_7, page=7)
Response(id=1_2_3_4_5_6_7_8, page=8)
Response(id=1_2_3_4_5_6_7_8_9, page=9)
Response(id=1_2_3_4_5_6_7_8_9_10, page=10)

我们如何使用 RxJava 进行分页 Web 服务调用，其中每个页面都依赖于前一个页面的响应，而无需递归？

How Can We Use RxJava for Paginated Web Service Calls, Where Each Page Depends on Previous Page Responses, Without Recursion?

rx-java

rx-java2