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

Question

我正在使用WebFlux实现一个RESTendpoint，我在基本操作上没有任何问题，但是有一个操作我不知道如何管理。我想返回与事件对象相关的PriceMessage对象，所以如果事件存在，endpoint返回一个ServerResponse.ok（），但是如果事件不存在，应该返回一个ServerResponse.not发现（）。

在存储库层中，如果存在eventId为的代码和事件，则有以下方法返回一个表示代码的Mono，并返回一个Mono。空（）如果没有：

@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;
}

然后，服务层返回一个流量，如果未找到事件或连接到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和map部件。

最后，在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

您可以在注册路由时使用“选择”来处理它。

@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();
}

或者您可以在服务器上使用用户onErrorResume：

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

希望这有帮助。

Answer 2

如果发生错误，可以使用OneErrorResume创建错误响应。

 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());