diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java index 431851d942..b6a9a99b06 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.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. @@ -88,7 +88,7 @@ public class WebFluxMessageHandlerSpec * @since 5.0.1 * @see WebFluxRequestExecutingMessageHandler#setBodyExtractor(BodyExtractor) */ - public WebFluxMessageHandlerSpec bodyExtractor(BodyExtractor bodyExtractor) { + public WebFluxMessageHandlerSpec bodyExtractor(BodyExtractor bodyExtractor) { this.target.setBodyExtractor(bodyExtractor); return this; } 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 bc4760d006..087602c0e3 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 @@ -23,7 +23,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.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.http.HttpEntity; @@ -73,7 +72,7 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx private boolean replyPayloadToFlux; - private BodyExtractor bodyExtractor; + private BodyExtractor bodyExtractor; private Expression publisherElementTypeExpression; @@ -166,7 +165,7 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx * @see #setExpectedResponseType(Class) * @see #setExpectedResponseTypeExpression(Expression) */ - public void setBodyExtractor(BodyExtractor bodyExtractor) { + public void setBodyExtractor(BodyExtractor bodyExtractor) { this.bodyExtractor = bodyExtractor; } @@ -208,71 +207,16 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx WebClient.RequestBodySpec requestSpec = createRequestBodySpec(uri, httpMethod, httpRequest, requestMessage, uriVariables); - Mono responseMono = exchangeForResponseMono(requestSpec); + Mono>> responseMono = exchangeForResponseMono(requestSpec, expectedResponseType); if (isExpectReply()) { - return createReplyFromResponse(expectedResponseType, responseMono); + return createReplyFromResponse(responseMono); } else { return responseMono.then(); } } - private Object createReplyFromResponse(Object expectedResponseType, Mono responseMono) { - return responseMono - .flatMap(response -> { - ResponseEntity.BodyBuilder httpEntityBuilder = - ResponseEntity.status(response.statusCode()) - .headers(response.headers().asHttpHeaders()); - - Mono bodyMono; - - if (expectedResponseType != null) { - if (this.replyPayloadToFlux) { - BodyExtractor, ReactiveHttpInputMessage> extractor; - if (expectedResponseType instanceof ParameterizedTypeReference) { - extractor = BodyExtractors.toFlux( - (ParameterizedTypeReference) expectedResponseType); - } - else { - extractor = BodyExtractors.toFlux((Class) expectedResponseType); - } - Flux flux = response.body(extractor); - bodyMono = Mono.just(flux); - } - else { - BodyExtractor, ReactiveHttpInputMessage> extractor; - if (expectedResponseType instanceof ParameterizedTypeReference) { - extractor = BodyExtractors.toMono( - (ParameterizedTypeReference) expectedResponseType); - } - else { - extractor = BodyExtractors.toMono((Class) expectedResponseType); - } - bodyMono = response.body(extractor); - } - } - else if (this.bodyExtractor != null) { - Object body = response.body(this.bodyExtractor); - if (body instanceof Mono) { - bodyMono = (Mono) body; - } - else { - bodyMono = Mono.just(body); - } - } - else { - bodyMono = Mono.empty(); - } - - return bodyMono - .map(httpEntityBuilder::body) - .defaultIfEmpty(httpEntityBuilder.build()); - } - ) - .map(this::getReply); - } - private WebClient.RequestBodySpec createRequestBodySpec(Object uri, HttpMethod httpMethod, HttpEntity httpRequest, Message requestMessage, Map uriVariables) { @@ -294,17 +238,6 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx return requestSpec; } - private Mono exchangeForResponseMono(WebClient.RequestBodySpec requestSpec) { - 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 private BodyInserter buildBodyInserterForRequest(Message requestMessage, HttpEntity httpRequest) { @@ -366,4 +299,75 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx } } + private Mono>> exchangeForResponseMono(WebClient.RequestBodySpec requestSpec, + Object expectedResponseType) { + + return requestSpec.retrieve() + .onStatus(HttpStatus::isError, ClientResponse::createException) + .toEntityFlux(createBodyExtractor(expectedResponseType)); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private BodyExtractor, ? super ClientHttpResponse> createBodyExtractor(Object expectedResponseType) { + if (expectedResponseType != null) { + if (this.replyPayloadToFlux) { + if (expectedResponseType instanceof ParameterizedTypeReference) { + return BodyExtractors.toFlux((ParameterizedTypeReference) expectedResponseType); + } + else { + return BodyExtractors.toFlux((Class) expectedResponseType); + } + } + else { + BodyExtractor, ReactiveHttpInputMessage> monoExtractor; + if (expectedResponseType instanceof ParameterizedTypeReference) { + monoExtractor = BodyExtractors.toMono((ParameterizedTypeReference) expectedResponseType); + } + else { + monoExtractor = BodyExtractors.toMono((Class) expectedResponseType); + } + return (inputMessage, context) -> Flux.from(monoExtractor.extract(inputMessage, context)); + } + } + else if (this.bodyExtractor != null) { + return (inputMessage, context) -> { + Object body = this.bodyExtractor.extract(inputMessage, context); + if (body instanceof Publisher) { + return Flux.from((Publisher) body); + } + return Flux.just(body); + }; + } + else { + return (inputMessage, context) -> Flux.empty(); + } + } + + private Object createReplyFromResponse(Mono>> responseMono) { + return responseMono + .flatMap(response -> { + ResponseEntity.BodyBuilder httpEntityBuilder = + ResponseEntity.status(response.getStatusCode()) + .headers(response.getHeaders()); + + Flux body = response.getBody(); + Mono bodyMono = Mono.empty(); + + if (body != null) { + if (this.replyPayloadToFlux) { + bodyMono = Mono.just(body); + } + else { + bodyMono = body.next(); + } + } + + return bodyMono + .map(httpEntityBuilder::body) + .defaultIfEmpty(httpEntityBuilder.build()); + } + ) + .map(this::getReply); + } + } 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 b247b39593..dbeef5b2d4 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 @@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.DataBufferLimitException; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.client.reactive.ClientHttpConnector; @@ -38,6 +39,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.test.web.reactive.server.HttpHandlerConnector; +import org.springframework.web.reactive.function.client.ExchangeStrategies; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClientResponseException; @@ -272,8 +274,8 @@ class WebFluxRequestExecutingMessageHandlerTests { assertThat(response.getHeaders().getContentType()).isEqualTo(MediaType.TEXT_PLAIN); StepVerifier.create( - response.getBody() - .map(dataBuffer -> new String(dataBuffer.asByteBuffer().array()))) + response.getBody() + .map(dataBuffer -> new String(dataBuffer.asByteBuffer().array()))) .expectNext("foo", "bar", "baz") .verifyComplete(); } @@ -289,14 +291,14 @@ class WebFluxRequestExecutingMessageHandlerTests { 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)) + 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) @@ -304,12 +306,12 @@ class WebFluxRequestExecutingMessageHandlerTests { }); WebClient webClient = WebClient.builder() - .clientConnector(httpConnector) - .build(); + .clientConnector(httpConnector) + .build(); String destinationUri = "https://www.springsource.org/spring-integration"; WebFluxRequestExecutingMessageHandler reactiveHandler = - new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); QueueChannel replyChannel = new QueueChannel(); QueueChannel errorChannel = new QueueChannel(); @@ -317,9 +319,9 @@ class WebFluxRequestExecutingMessageHandlerTests { reactiveHandler.setBodyExtractor(new ClientHttpResponseBodyExtractor()); final Message message = - MessageBuilder.withPayload("hello, world") - .setErrorChannel(errorChannel) - .build(); + MessageBuilder.withPayload("hello, world") + .setErrorChannel(errorChannel) + .build(); reactiveHandler.handleMessage(message); Message errorMessage = errorChannel.receive(10_000); @@ -331,4 +333,49 @@ class WebFluxRequestExecutingMessageHandlerTests { assertThat(throwable.getCause()).isInstanceOf(WebClientResponseException.NotFound.class); assertThat(throwable.getMessage()).contains("404 Not Found"); } + + @Test + void testMaxInMemorySizeExceeded() { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + response.setStatusCode(HttpStatus.OK); + + DataBufferFactory bufferFactory = response.bufferFactory(); + + Mono data = Mono.just(bufferFactory.wrap("test".getBytes())); + + return response.writeWith(data) + .then(Mono.defer(response::setComplete)); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .exchangeStrategies(ExchangeStrategies.builder() + .codecs(clientCodecConfigurer -> clientCodecConfigurer + .defaultCodecs() + .maxInMemorySize(1)) + .build()) + .build(); + + String destinationUri = "https://www.springsource.org/spring-integration"; + WebFluxRequestExecutingMessageHandler reactiveHandler = + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); + + reactiveHandler.setExpectedResponseType(String.class); + + QueueChannel errorChannel = new QueueChannel(); + reactiveHandler.handleMessage(MessageBuilder.withPayload("").setErrorChannel(errorChannel).build()); + + Message errorMessage = errorChannel.receive(10000); + assertThat(errorMessage).isNotNull(); + + Object payload = errorMessage.getPayload(); + assertThat(payload).isInstanceOf(MessageHandlingException.class) + .extracting("cause") + .isInstanceOf(WebClientResponseException.class) + .extracting("cause") + .isInstanceOf(DataBufferLimitException.class) + .extracting("message") + .isEqualTo("Exceeded limit on max bytes to buffer : 1"); + } + }