如果 Flux 流数据有错误状态，我如何返回不同的 ServerResponse

Question

我正在使用 WebFlux 实现一个 REST 端点，我在处理基本操作时没有任何问题，但是有一个操作我不知道如何管理。 我想返回与 Event 对象相关的 PriceMessage 对象，因此如果事件存在，则端点返回 ServerResponse.ok() 但如果 Event 不存在，则应返回 ServerResponse.notFound()。

在存储库层中有以下方法，如果存在则返回表示代码的 Mono 和带有 eventId 的事件，如果不存在则返回 Mono.empty()：

@Override
public Mono<Integer> getSportsEventId(long eventId) {
  Optional<SourceEventDto> optionalSourceEvent =
    springJpaSourceEventRepository.findByEventId(eventId);

  Mono<Integer> result = Mono.empty();

  if (optionalSourceEvent.isPresent()) {
    result = Mono.just(new Integer(optionalSourceEvent.get().getSourceId()));
  }

  return result;
}

然后，服务层返回一个 Flux，如果事件未找到或连接到 Redis 以获取与 sportsEventId 相关的价格，则设置错误：

@Override
public Flux<PriceMessage> getPrices(long eventId) {
  return
    // get the sportsEventId
    sourceEventRepository.getSportsEventId(eventId)
      // notify the event does not exist
      .switchIfEmpty(Mono.error(new IllegalStateException("Event " + eventId + " does not exist")))
      // get the related PriceEntity objects
      .flatMapMany(priceRepository::findBySporsEventId)
      // transform to PriceMessage
      .map(priceMessageFactory::from);
}

此代码经过单元测试，我还对其进行了调试，以确保在流中设置错误时不会调用 flatMapMany 和地图部分。

最后，在 REST 层有这段代码，幸福之路：

public Mono<ServerResponse> getPricesByEventId(ServerRequest request) {
  String eventIdParam = request.pathVariable("eventId");

  // call the service layer
  Flux<PriceMessage> prices = priceService.getPrices(eventId);

  return
    ServerResponse.ok()
      .contentType(MediaType.APPLICATION_STREAM_JSON)
      .body(prices, PriceMessage.class);
}

这工作正常，但如果价格流包含错误，因为未找到事件，我不知道如何返回 404 错误，在正文中设置错误消息，如下所示：

ServerResponse.status(HttpStatus.NOT_FOUND)
  .contentType(MediaType.APPLICATION_JSON_UTF8)
  .syncBody(<error_message_from_flux>);

伪代码是这样的，但我无法将其转换为响应式：

if (!prices.isError()) then
  return
    ServerResponse.ok()
      .contentType(MediaType.APPLICATION_STREAM_JSON)
      .body(prices, PriceMessage.class);
else
  return
    ServerResponse.status(HttpStatus.NOT_FOUND)
     .contentType(MediaType.APPLICATION_JSON_UTF8)
     .syncBody(<error_message_from_flux>);
end

任何帮助将不胜感激。

Answer 1

您可以使用 onErrorResume 创建错误响应以防发生错误。

 return priceService.getPrices(eventId).flatMap(prices -> ServerResponse.ok()
          .contentType(MediaType.APPLICATION_STREAM_JSON)
          .body(prices, PriceMessage.class))
       .onErrorResume(err -> ServerResponse.status(HttpStatus.NOT_FOUND).body({Your Error Body Here}).build());

Answer 2

注册路由时可以使用choose来处理。

@Bean
public RouterFunction<ServerResponse> route(PriceMessageHandler handler) {
  return RouterFunctions.route().GET("/",
      request -> ServerResponse.ok().body(handler.handle(request), PriceMessage.class))
      .onError(Exception.class, (e, a) -> ServerResponse.status(HttpStatus.NOT_FOUND).build())
      .build();
}

或者您可以在 ServerResponse 上使用 onErrorResume：

return ServerResponse.ok().body(priceService.getPrices(eventId), PriceMessage.class).onErrorResume((e) -> ServerResponse.notFound().build());

希望这会有所帮助。