在 Flux.error 的情况下返回相关的 ServerResponse答案

【问题标题】：Return relevant ServerResponse in case of Flux.error在 Flux.error 的情况下返回相关的 ServerResponse
【发布时间】：2022-06-30 13:41:05
【问题描述】：

我有一个 WebFlux 功能性 REST 端点，但由于我的代码中引发异常（例如无效路径变量上的 BadRequest），我无法返回自定义 http 错误。

考虑一下我的处理程序：

public Mono<ServerResponse> getStarships(ServerRequest request) {
    String starshipType = request.pathVariable("type");
    return ServerResponse
            .ok()
            .contentType(APPLICATION_JSON)
            .body(starshipService.getFromSpacedock(starshipType), Starship.class)
            .onErrorResume(InvalidStarshipTypeException.class, 
                           e -> ServerResponse
                                  .badRequest()
                                  .bodyValue(e.getMessage()));
}

当starshipService.getFromSpacedock(starshipType) 返回Flux.just(new Starship()) 时，一切正常。

当它返回 Flux.error(new InvalidStarshipTypeException("invalid starship type")) 时，我希望 onErrorResume 启动并返回我的自定义 BadRequest ServerResponse 和我的消息。

相反，我的端点以 500 响应（其中包含我的自定义异常）。 onErrorResume 被忽略。

我该如何解决这个问题？

我尝试过的：

将异常包装在ResponseStatusException 中：我得到了400，但不是通过自定义ServerResponse 路由。这种方法的问题是我必须将 Spring 配置为在以这种方式处理异常时显示消息，这是我不希望的。
在 Flux 上使用 flatMap，但这会导致 Flux<ServerResponse> 而不是 Mono<ServerResponse>：

return starshipService.getFromSpacedock(starshipType) // remember, this is a Flux<Starship>
    .flatMap(ships -> ServerResponse.ok()
                          .contentType(MediaType.APPLICATION_JSON)
                          .body(ships, StarShip.class))
    .onErrorResume(e -> ServerResponse.badRequest().bodyValue(e.getMessage()));

【问题讨论】：

I would have to add some global exception handling like you would with annotation style 然后使用注释样式的端点。您已选择使用functional styled 端点，它们被认为是较低级别，这意味着您可以处理自己的异常并返回自己的响应。
@Toerktumlare 这就是我试图使用 onErrorResume 完成的任务 - 关于为什么这不起作用的任何想法？我错过了什么？
您的实际问题是什么？您所写的只是这一切似乎都可以工作It would seem that the onErrorResume, as I have put it in the code, would do the trick 请更新并非常清楚，您希望它如何工作，它现在如何工作，使用什么请求，因为这不清楚。
@Toerktumlare 为复活节造成的延误表示歉意。我试图澄清我的问题。请注意，stackoverflow.com/questions/58429966/… 和 stackoverflow.com/questions/64578647/… 都没有回答我的问题，尽管有 reactor 和 spring 文档的链接

标签： java spring-webflux

【解决方案1】：

我遇到了类似的问题。要解决此问题，您需要首先将冷焊剂转换为热焊剂。然后在热通量上调用.next()，返回一个Mono<Starship>。在此单声道上，请致电 .flatMap().switchIfEmpty().onErrorResume()。在flatMap() 中将返回的startship 对象与热通量连接起来。

这是修改后的代码 sn-p 来实现你想要的：

public Mono&lt;ServerResponse&gt; getStarships(ServerRequest request) 
{
String starshipType = request.pathVariable("type");

Flux&lt;Starship&gt; coldStarshipFlux = starshipService.getFromSpacedock(starshipType);

//The following step is a very important step. It converts your cold flux into a hot flux.
Flux&lt;Startship&gt; hotStarshipFlux = coldStarshipFlux.publish().refCount(1, Duration.ofSeconds(2));

return hotStarshipFlux.next()
                    .flatMap( starShipObj ->
                       {
                            Flux&lt;Starship&gt; flux = Mono.just(starShipObj)
                                                        .concatWith(hotStarshipFlux);
                            
                            return ServerResponse.ok()
                                            .contentType(MediaType.APPLICATION_JSON)
                                             .body(flux, Starship.class);
                        }
                    )
                    .switchIfEmpty(
                        ServerResponse.notFound().build()
                    )
                    .onErrorResume( t ->
                        {
                            if(t instanceof InvalidStarshipTypeException)
                            {
                                return ServerResponse.badRequest()
                                                    .contentType(MediaType.TEXT_PLAIN)
                                                    .bodyValue(t.getMessage());
                            }
                            else
                            {
                                return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
                                                        .contentType(MediaType.TEXT_PLAIN)
                                                       .bodyValue(t.getMessage());
                            }
                        });
}

代码.publish().refCount(1, Duration.ofSeconds(2)); 使您的冷焊剂变成热焊剂。这样做很重要。

使用热通量时，每个新订阅共享通量流中发出的元素，这些元素在订阅开始后发出。因此，对.next() 的初始调用导致热通量发射第一个元素。现在，当在同一个热通量上再次调用.concatWith() 方法时，热通量不会再次重新发射第一个元素，而是继续发射其流中的后续元素。所以流中所有剩余的元素，从第二个开始将被连接起来。

如果您没有将助焊剂转换为热助焊剂，而是在冷助焊剂上运行上述代码，那么对 .next() 和 .concatWith() 的调用都会导致冷助焊剂重新生成相同的数据两次，一次用于.next()，一次用于.concatWith()。因此，您将有一个重复的第一个元素。

现在，您可能会问，为什么要为所有这些忽冷忽热的流量烦恼？为什么不只使用 cold 助焊剂来执行以下代码 sn-p 之类的操作？毕竟感冒会重新生成数据，所以根本不需要调用concatWith() 方法。

Flux&lt;Starship&gt; coldStarshipFlux = starshipService.getFromSpacedock(starshipType);

return coldStarldshipFlux.next()
                         .flatMap( starShipObj -> // ignore the startShipObj
                            {
                                return ServerResponse.ok()
                                                 .contentType(MediaType.APPLICATION_JSON)
                                                 .body(coldStarldshipFlux, Starship.class);
                            }
                        )
                        .switchIfEmpty(
                             ... //same code as above hot flux
                        )
                        .onErrorResume( t ->
                            {
                                ... //same code as as above hot flux
                            });

上述代码 sn-p 的问题是冷通量上的所有订阅都重新生成数据，每个订阅。因此，有效地调用.next() 和.concatWith() 将导致冷通量重新生成相同的数据流，并且取决于您的 startshipService 的编码方式，可能会导致第二个 HTTP提出请求。因此，实际上，您将发出两个 HTTP 请求而不是一个。

使用热通量时，需要重新生成数据，从而避免潜在地发出第二个 HTTP 请求。这是将冷焊剂转化为热焊剂的最大优势。

有关冷热助焊剂的详细信息，请参阅以下网站。最后，它非常清楚地解释了热通量和冷通量之间的区别，以及它们的行为方式有何不同：
https://spring.io/blog/2019/03/06/flight-of-the-flux-1-assembly-vs-subscription

希望这个回答对你有帮助。

【讨论】：