kotlin - 在 Kotlin 中使用 Reactor 分页 Gitlab API

Question

我正在编写一段 Kotlin 代码，该代码使用 react 器框架来实现对 Gitlab 提交 API 的调用。
提交 API 是分页的。我正在努力检索的函数“在”两个指定的提交哈希之间检索提交。

只要它实际上可以检索任何提交，该函数就可以正常工作，但如果找不到结果则失败。然后它失败了java.lang.RuntimeException: Reached end of commit log .

我尝试替换行 .switchIfEmpty(Flux.error(RuntimeException("Reached end of commit log.")))与 .switchIfEmpty(Flux.empty()) ，但这会产生一个无限循环。

我不太了解多个通量的嵌套，这让我很难调试。我非常感谢有关如何解决此问题的任何提示。

fun getCommits(fromCommit: String, toCommit: String): Iterable<Commit> {
    val commits = Flux.concat(Flux.generate<Flux<GitLabCommit>, Int>({ 1 }) { state, sink ->
        val page = client.get()
                .uri("/projects/{name}/repository/commits?page=$state&per_page=100")
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToFlux<GitLabCommit>()
                .doOnError({
                    LOGGER.warn("Could not retrieve commits for project '$name': ${it.message}")
                    sink.next(Flux.just(GitLabCommit("xxxxx", "Could not retrieve all commits for project '$name'")))
                    sink.complete()
                })
                .onErrorReturn(GitLabCommit("xxxxx", "Could not retrieve all commits for project '$name'"))
                .switchIfEmpty(Flux.error(RuntimeException("Reached end of commit log.")))

        sink.next(page)
        return@generate state + 1
    })


    return commits
            // The Gitlab API returns commits from newest to oldest
            .skipWhile { !it.id.startsWith(toCommit) } //inclusive
            .takeWhile { !it.id.startsWith(fromCommit) } //exclusive
            .map { Commit(it.title, listOf(it.id), name) }
            .toIterable()
}

关于上述代码的其他提示:

这是 GitlabCommit 类:

@JsonIgnoreProperties(ignoreUnknown = true)
private data class GitLabCommit(val id: String, val title: String)

client是 org.springframework.web.reactive.function.client.WebClient.Builder 的正确初始化实例，这有助于 token 处理和 URL 编码。

Answer 1

异常来源:您对 commits 的外部使用没有 onErrorResume条款。
无限循环的来源:内部generate中没有错误，它将继续循环，增加state并连接一个空结果。

我会采用与生成不同的方法:range + concatMap + takeWhile。

像这样的东西:

fun getCommits(fromCommit: String, toCommit: String): Iterable<String> =
    Flux.range(1, 1000) //tune the second parameter
            .concatMap { page -> client
                .get()
                .uri("/projects/{name}/repository/commits?page=$page&per_page=100")
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToFlux<GitLabCommit>()
                .doOnError({ LOGGER.warn("Could not retrieve commits for project '$name': ${it.message}")})
                .onErrorReturn(GitLabCommit("xxxxx", "Could not retrieve all commits for project '$name'"))
                .defaultIfEmpty(GitLabCommit("xxxxx", "Reached end of commit log."))
                // ^ we now have marker commits in case of error or end of log
            }
            //use the marker commits to short-circuit the range
            .takeWhile { !it.id.equals("xxxxx") } //end condition to cancel the range loop, doesn't include the xxxxx commits
            // The Gitlab API returns commits from newest to oldest
            .skipWhile { !it.id.startsWith(toCommit) } //inclusive
            .takeWhile { !it.id.startsWith(fromCommit) } //exclusive
            .map { Commit(it.title, listOf(it.id), name) }
            .toIterable()

我们从最多 1000 个页面的范围开始，为每个页面发出请求。

这可能会导致无关的请求(例如，如果到第 100 页我们已经收到一个空响应，则不会有更多页面)。你几乎解决了这个问题:

我们可以同时使用 onErrorReturn和 defaultIfEmpty创建一个标记提交，然后可以在 concatMap 之后使用在 takeWhile .

一旦额外的 takeWhile看到第一个 xxxxx提交，它将触发将传播到 concatMap/range 的取消，导致范围停止发出页码，从而停止发出无关请求。

然后你有你的业务逻辑 skipWhile , takeWhile和 map .