From c70c98b945d731d18efb6c763d6132a82cffb9eb 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`** # Conflicts: # spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java # spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java --- ...WebFluxRequestExecutingMessageHandler.java | 47 +++----------- .../webflux/dsl/WebFluxDslTests.java | 7 +-- ...uxRequestExecutingMessageHandlerTests.java | 62 ++++++++++++++++++- 3 files changed, 73 insertions(+), 43 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 2c34c7fa2e..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 * @@ -298,38 +295,14 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx } private Mono exchangeForResponseMono(WebClient.RequestBodySpec requestSpec) { - return requestSpec.exchange() - .flatMap(response -> { - HttpStatus httpStatus = response.statusCode(); - if (httpStatus.isError()) { - 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)); - } - ); - } - else { - return Mono.just(response); - } - }); + return requestSpec.retrieve() + .onStatus(HttpStatus::isError, ClientResponse::createException) + .toEntityList(DataBuffer.class) + .map((entity) -> + ClientResponse.create(entity.getStatusCode()) + .headers((headers) -> headers.addAll(entity.getHeaders())) + .body(Flux.fromIterable(entity.getBody())) // NOSONAR - not null according toEntityList() + .build()); } @Nullable diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java index 5e0cd1d46d..1259df6f45 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-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. @@ -25,8 +25,6 @@ import java.security.Principal; import java.time.Duration; import java.util.Collections; -import javax.annotation.Resource; - import org.hamcrest.Matchers; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -118,7 +116,8 @@ public class WebFluxDslTests { @Qualifier("webFluxWithReplyPayloadToFlux.handler") private WebFluxRequestExecutingMessageHandler webFluxWithReplyPayloadToFlux; - @Resource(name = "httpReactiveProxyFlow.webflux:outbound-gateway#0") + @Autowired + @Qualifier("httpReactiveProxyFlow.webflux:outbound-gateway#0") private WebFluxRequestExecutingMessageHandler httpReactiveProxyFlow; @Autowired 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 129ccacfb9..0bccd75b24 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-2019 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"); + } }