Spring Webflux ： Webclient ： Get body on error

java spring reactor spring-webflux

2022-09-01 01:45:32

我正在使用来自spring webflux的webclient，像这样：

WebClient.create()
            .post()
            .uri(url)
            .syncBody(body)
            .accept(MediaType.APPLICATION_JSON)
            .headers(headers)
            .exchange()
            .flatMap(clientResponse -> clientResponse.bodyToMono(tClass));

它运行良好。我现在想要处理来自我正在调用的Web服务的错误（例如500内部错误）。通常我会在“流”上添加一个doOnError，并isu the Throwable来测试状态代码，

但我的问题是，我想获得Web服务提供的主体，因为它为我提供了一条我想使用的消息。

无论发生什么，我都希望做flatMap，并测试自己的状态代码以反序列化或不反序列化正文。

答案 1

我更喜欢使用 ClientResponse 提供的方法来处理 http 错误并引发异常：

WebClient.create()
         .post()
         .uri( url )
         .body( bodyObject == null ? null : BodyInserters.fromValue( bodyObject ) )
         .accept( MediaType.APPLICATION_JSON )
         .headers( headers )
         .exchange()
         .flatMap( clientResponse -> {
             //Error handling
             if ( clientResponse.statusCode().isError() ) { // or clientResponse.statusCode().value() >= 400
                 return clientResponse.createException().flatMap( Mono::error );
             }
             return clientResponse.bodyToMono( clazz )
         } )
         //You can do your checks: doOnError (..), onErrorReturn (..) ...
         ...

实际上，它与 DefaultWebClient 的 DefaultResponseSpec 中用于处理错误的逻辑相同。DefaultResponseSpec 是 ResponseSpec 的实现，如果我们创建 retrieve（）而不是 exchange（），我们将拥有它。

答案 2

难道我们没有吗？onStatus()

    public Mono<Void> cancel(SomeDTO requestDto) {
        return webClient.post().uri(SOME_URL)
                .body(fromObject(requestDto))
                .header("API_KEY", properties.getApiKey())
                .retrieve()
                .onStatus(HttpStatus::isError, response -> {
                    logTraceResponse(log, response);
                    return Mono.error(new IllegalStateException(
                            String.format("Failed! %s", requestDto.getCartId())
                    ));
                })
                .bodyToMono(Void.class)
                .timeout(timeout);
    }

和：

    public static void logTraceResponse(Logger log, ClientResponse response) {
        if (log.isTraceEnabled()) {
            log.trace("Response status: {}", response.statusCode());
            log.trace("Response headers: {}", response.headers().asHttpHeaders());
            response.bodyToMono(String.class)
                    .publishOn(Schedulers.elastic())
                    .subscribe(body -> log.trace("Response body: {}", body));
        }
    }