使用RSocket和Project Reactor实现202接受后重试行为

Question

我正在实现一个典型的用例，其中客户端请求异步生成的资源。因此,立即生成并返回resourceID：

1. CLIENT ---(POST /request-resource)--->  SERVER
2. SERVER (Generates resID, launches async process) ---(202 Accepted - resID)---> CLIENT

此时，服务器中有一个后台任务，该任务将最终生成一个结果，并将其存储在与RESID相关联的数据库中。客户端将定期请求资源,重试直到资源可用：

3. CLIENT ---(/resource/resID)--->  SERVER (checks in Postgres using reactive driver)
4. SERVER ---(404 - Retry-After 5)---> CLIENT
5. CLIENT ---(/resource/resID)--->  SERVER (checks in Postgres using reactive driver)
6. SERVER ---(200 - JSON Payload)---> CLIENT

我认为RSocket是一个完美的选择，以避免在资源可用之前进行无休止的客户端重试（步骤3.on）。

哪种交互模型会更适合这个问题，我如何实现它？

考虑存储库如下:ResourceRepository.mono getResult(String resID)

如果我选择了请求/响应交互模型，那么我的情况与前面一样。除非有办法让一个单声道重试直到有结果。这可能吗？

对于Request/Stream，我可以使用response.status=processing返回类似Flux 的结果，直到对Postgre的查询返回一个结果，然后Flux将有一个response.status=ok的元素，并且Flux将完成。需要一个最大的时间来完成在配置的周期内没有结果的通量。既然如此，我怎么能这样做呢？

我需要创建一个Flux，该Flux周期性地发出（具有最大周期超时），在存储库返回空Mono时有一个没有结果的元素，或者在te存储库有它时有一个实际值，完成Flux。

Answer 1

此问题的解决方案使用RSocket和RequestResponse交互模型，该交互模型等待资源在DB中可用。关键是使用repeatwhenempty运算符：

    @MessageMapping("request-resource")
    fun getResourceWebSocket(resourceRequest: ResourceRequest): Mono<Resource> {
        return resourceService.sendResourceRequestProcessing(resourceRequest)
    }

    override fun sendResourceRequestMessage(resourceRequest: ResourceRequest): Mono<Resource> {
        val resourceId = randomUUID().toString()
        return Mono.fromCallable {
            sendKafkaResourceProcessingRequestMessage(resourceId, resourceRequest)
        }.then(poolResourceResponse(resourceId))
    }

    private fun poolResourceResponse(resourceId: String): Mono<Resource> {
        return resourceRepository.findByResourceId(resourceId)
                .repeatWhenEmpty(30) { longFlux ->
                    longFlux.delayElements(Duration.ofSeconds(1))
                            .doOnNext { logger.info("Repeating {}", it) }
                }
    }