Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3610: Fix WebFluxMH for error handling

Browse Source 
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
This commit is contained in:
Dave G
2021-08-09 13:27:12 -04:00
committed by Artem Bilan
parent 125ac454f5
commit c70c98b945
3 changed files with 73 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java
View File

@@ -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<ClientResponse> 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

7
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java
View File

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

62
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java
View File

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