From 1a73151114dd97e4659f8cd594e4c489c30cbc09 Mon Sep 17 00:00:00 2001 From: Dave G Date: Mon, 9 Aug 2021 13:27:12 -0400 Subject: [PATCH] GH-3610: Fix WebFluxMH for error handling Fixes https://github.com/spring-projects/spring-integration/issues/3610 This fix changes the WebFluxRequestExecutingMessageHandler The change specifically changes the contruction of the WebClientResponseException to use the `create` factory method. * Added changes and unit test to cover updated code Updated code to use simplified exception construction that is more tolerant of larger payloads. Updated unit tests to check for specific exception types. * Corrected checkstyle errors Corrected checkstyle error in imports. Corrected whitespace between casts. Corrected trailing whitespace. * Added @author **Cherry-pick to `5.4.x` & `5.3.x`** --- ...WebFluxRequestExecutingMessageHandler.java | 34 +--------- ...uxRequestExecutingMessageHandlerTests.java | 62 ++++++++++++++++++- 2 files changed, 63 insertions(+), 33 deletions(-) diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java index 66c1889e33..bc4760d006 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2020 the original author or authors. + * Copyright 2017-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,7 +17,6 @@ package org.springframework.integration.webflux.outbound; import java.net.URI; -import java.nio.charset.StandardCharsets; import java.util.Map; import org.reactivestreams.Publisher; @@ -25,7 +24,6 @@ import org.reactivestreams.Publisher; import org.springframework.core.ParameterizedTypeReference; import org.springframework.core.io.Resource; import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.http.HttpEntity; @@ -41,7 +39,6 @@ import org.springframework.integration.http.outbound.AbstractHttpRequestExecutin import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; -import org.springframework.util.MimeType; import org.springframework.util.MultiValueMap; import org.springframework.web.reactive.function.BodyExtractor; import org.springframework.web.reactive.function.BodyExtractors; @@ -49,7 +46,6 @@ import org.springframework.web.reactive.function.BodyInserter; import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.reactive.function.client.WebClientResponseException; import org.springframework.web.util.DefaultUriBuilderFactory; import reactor.core.publisher.Flux; @@ -62,6 +58,7 @@ import reactor.core.publisher.Mono; * @author Shiliang Li * @author Artem Bilan * @author Gary Russell + * @author David Graff * * @since 5.0 * @@ -299,32 +296,7 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx private Mono exchangeForResponseMono(WebClient.RequestBodySpec requestSpec) { return requestSpec.retrieve() - .onStatus(HttpStatus::isError, response -> { - HttpStatus httpStatus = response.statusCode(); - return response.body(BodyExtractors.toDataBuffers()) - .reduce(DataBuffer::write) - .map(dataBuffer -> { - byte[] bytes = new byte[dataBuffer.readableByteCount()]; - dataBuffer.read(bytes); - DataBufferUtils.release(dataBuffer); - return bytes; - }) - .defaultIfEmpty(new byte[0]) - .map(bodyBytes -> { - throw new WebClientResponseException( - "ClientResponse has erroneous status code: " - + httpStatus.value() + " " - + httpStatus.getReasonPhrase(), - httpStatus.value(), - httpStatus.getReasonPhrase(), - response.headers().asHttpHeaders(), - bodyBytes, - response.headers().contentType() - .map(MimeType::getCharset) - .orElse(StandardCharsets.ISO_8859_1)); - } - ); - }) + .onStatus(HttpStatus::isError, ClientResponse::createException) .toEntityList(DataBuffer.class) .map((entity) -> ClientResponse.create(entity.getStatusCode()) diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java index 6d5ccab8f7..b247b39593 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2020 the original author or authors. + * Copyright 2017-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,7 @@ package org.springframework.integration.webflux.outbound; import static org.assertj.core.api.Assertions.assertThat; +import java.nio.charset.StandardCharsets; import java.time.Duration; import org.junit.jupiter.api.Test; @@ -47,6 +48,7 @@ import reactor.test.StepVerifier; /** * @author Shiliang Li * @author Artem Bilan + * @author David Graff * * @since 5.0 */ @@ -108,6 +110,8 @@ class WebFluxRequestExecutingMessageHandlerTests { assertThat(errorMessage).isNotNull(); assertThat(errorMessage).isInstanceOf(ErrorMessage.class); Throwable throwable = (Throwable) errorMessage.getPayload(); + assertThat(throwable).isInstanceOf(MessageHandlingException.class); + assertThat(throwable.getCause()).isInstanceOf(WebClientResponseException.Unauthorized.class); assertThat(throwable.getMessage()).contains("401 Unauthorized"); } @@ -173,7 +177,8 @@ class WebFluxRequestExecutingMessageHandlerTests { assertThat(payload).isInstanceOf(MessageHandlingException.class); Exception exception = (Exception) payload; - assertThat(exception.getCause()).isInstanceOf(WebClientResponseException.class); + assertThat(exception).isInstanceOf(MessageHandlingException.class); + assertThat(exception.getCause()).isInstanceOf(WebClientResponseException.ServiceUnavailable.class); assertThat(exception.getMessage()).contains("503 Service Unavailable"); Message replyMessage = errorChannel.receive(10); @@ -273,4 +278,57 @@ class WebFluxRequestExecutingMessageHandlerTests { .verifyComplete(); } + + @Test + void testClientHttpResponseErrorAsReply() { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + response.setStatusCode(HttpStatus.NOT_FOUND); + response.getHeaders().setContentType(MediaType.APPLICATION_JSON); + + DataBufferFactory bufferFactory = response.bufferFactory(); + + Flux data = + Flux.just( + bufferFactory.wrap("{".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap(" \"error\": \"Not Found\",".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap(" \"message\": \"404 NOT_FOUND\",".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap(" \"path\": \"/spring-integration\",".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap(" \"status\": 404,".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap(" \"timestamp\": \"1970-01-01T00:00:00.000+00:00\",".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap(" \"trace\": \"some really\nlong\ntrace\",".getBytes(StandardCharsets.UTF_8)), + bufferFactory.wrap("}".getBytes(StandardCharsets.UTF_8)) + ); + + return response.writeWith(data) + .then(Mono.defer(response::setComplete)); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + String destinationUri = "https://www.springsource.org/spring-integration"; + WebFluxRequestExecutingMessageHandler reactiveHandler = + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); + + QueueChannel replyChannel = new QueueChannel(); + QueueChannel errorChannel = new QueueChannel(); + reactiveHandler.setOutputChannel(replyChannel); + reactiveHandler.setBodyExtractor(new ClientHttpResponseBodyExtractor()); + + final Message message = + MessageBuilder.withPayload("hello, world") + .setErrorChannel(errorChannel) + .build(); + reactiveHandler.handleMessage(message); + + Message errorMessage = errorChannel.receive(10_000); + + assertThat(errorMessage).isNotNull(); + assertThat(errorMessage).isInstanceOf(ErrorMessage.class); + final Throwable throwable = (Throwable) errorMessage.getPayload(); + assertThat(throwable).isInstanceOf(MessageHandlingException.class); + assertThat(throwable.getCause()).isInstanceOf(WebClientResponseException.NotFound.class); + assertThat(throwable.getMessage()).contains("404 Not Found"); + } }