project-reactor 相关问题

我正在使用 RxJava 2 通过 micronaut 框架进行反应式编程。我试图了解以下代码的问题：有趣的 getItemDetails( itemRequest：请求到 ): 单声道我正在使用 RxJava 2 通过 micronaut 框架进行反应式编程。我试图理解以下代码的问题： fun getItemDetails( itemRequest: RequestTo ): Mono<List<ResponseTo>> { return itemRequest.itemIdList.toFlux() .flatMap { findItemStatus(it) } .flatMap { response -> Mono.justOrEmpty(ResponseTo(response)) .onErrorResume { logger.error("Error occurred while mapping entity to response: $response", it) Mono.empty() } } .collectList() } RequestTo：由一个String列表组成。这个想法是迭代这个列表，从数据库中获取每个项目的详细信息（通过返回数据库实体的findItemStatus()函数调用处理）并将其映射到响应对象。我想这样工作，如果 findItemStatus() flatMap 操作返回的任何实体在将其映射到 ResponseTo 时导致异常，我希望在最终列表中跳过该项目但继续处理通量中的剩余项目.我不想将 onErrorResume 块保留在第二个 flatMap 之外，因为如果通量中的任何中间实体在映射时导致错误并且通量中的其余项目将不会被映射，这将关闭流。我不确定我的代码有什么问题，但是当 Mono.justOrEmpty(ResponseTo(response)) 中抛出异常时，它不会在 onErrorResume 中被捕获。我把测试用例放在这里： def "test getItemDetails() when there is an error mapping DB response to Response TO for any item"() { def itemEntity1 = new DiscontinuedItemsEntity("52782904", null, null, UUID.randomUUID()) def itemEntity2 = new DiscontinuedItemsEntity("52778904", null, null, UUID.randomUUID()) def requestBody = new DiscontinuedItemsTcinRequest(["52782904", "52782904"]) when: def response = itemsStatusService.getItemDetails(requestBody).block() def responseList = response.itemsList then: 1 * discontinuedItemsRepository.findDiscontinuedItems(*_) >> Mono.just(itemEntity1) 1 * discontinuedItemsRepository.findDiscontinuedItems(*_) >> Mono.just(itemEntity2) responseList.size() == 1 responseList.get(0).itemId == itemEntity1.itemId } 我需要这个测试用例通过，但是异常被抛出并且没有被处理，因为它没有被 onErrorResume 捕获

java reactive-programming project-reactor flux

回答 0 投票 0

如何理解Reactor中的Flux.sampleTimeout方法？

此方法的 2 个重载： public final Flux sampleTimeout(Function> throttlerFactory) 和公开决赛 Flux

java spring-webflux reactive-programming project-reactor reactive

回答 0 投票 0

Project Reactor：retryWhen() 的超时未按预期工作

下面的代码使用project reactor来模拟一个简单的交易，我们首先使用newTrx()来启动一个trx，然后调用getTrxResult()来获取trx的结果。请注意，我们将重试 doGet() s ...

project-reactor

回答 0 投票 0

Flux with repeatWhen 使用 onCancel 永远不会终止

我想用我的代码做什么我想构建一个无限的 Flux 来检查条件（功能标志），然后根据条件是否为真继续处理请求。 T...