diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java index 0da8b66a1b..ea8ec03797 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java @@ -21,6 +21,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.http.config.HttpOutboundGatewayParser; import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler; import org.springframework.util.StringUtils; @@ -46,6 +47,7 @@ public class WebFluxOutboundGatewayParser extends HttpOutboundGatewayParser { .addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef)); } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-to-flux"); return builder; } 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 f181cbb3f3..29e800000c 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 @@ -29,6 +29,7 @@ import org.springframework.expression.common.LiteralExpression; import org.springframework.http.HttpEntity; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; +import org.springframework.http.ReactiveHttpInputMessage; import org.springframework.http.ResponseEntity; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.http.outbound.AbstractHttpRequestExecutingMessageHandler; @@ -36,12 +37,14 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; import org.springframework.util.MimeType; +import org.springframework.web.reactive.function.BodyExtractor; import org.springframework.web.reactive.function.BodyExtractors; 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 reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** @@ -59,6 +62,8 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx private final WebClient webClient; + private boolean replyToFlux; + /** * Create a handler that will send requests to the provided URI. * @param uri The URI. @@ -110,6 +115,20 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx this.setAsync(true); } + /** + * The boolean flag to identify if the reply payload should be as a {@link Flux} from the response body + * or as resolved value from the {@link Mono} of the response body. + * Defaults to {@code false} - simple value is pushed downstream. + * Makes sense when {@code expectedResponseType} is configured. + * @param replyToFlux represent reply payload as a {@link Flux} or as a value from the {@link Mono}. + * @since 5.0.1 + * @see #setExpectedResponseType(Class) + * @see #setExpectedResponseTypeExpression(Expression) + */ + public void setReplyToFlux(boolean replyToFlux) { + this.replyToFlux = replyToFlux; + } + @Override public String getComponentType() { return (isExpectReply() ? "webflux:outbound-gateway" : "webflux:outbound-channel-adapter"); @@ -169,13 +188,35 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx ResponseEntity.status(response.statusCode()) .headers(response.headers().asHttpHeaders()); - Mono bodyMono = Mono.empty(); + Mono bodyMono; - if (expectedResponseType instanceof ParameterizedTypeReference) { - bodyMono = response.body(BodyExtractors.toMono((ParameterizedTypeReference) expectedResponseType)); + if (expectedResponseType != null) { + if (this.replyToFlux) { + 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 (expectedResponseType != null) { - bodyMono = response.body(BodyExtractors.toMono((Class) expectedResponseType)); + else { + bodyMono = Mono.empty(); } return bodyMono diff --git a/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd b/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd index 4397635b54..b9f8660cfe 100644 --- a/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd +++ b/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd @@ -10,7 +10,7 @@ + schemaLocation="http://www.springframework.org/schema/integration/spring-integration-5.0.xsd"/> @@ -82,7 +82,8 @@ - + Specify an expression for URI variable placeholder within 'url'. @@ -126,7 +127,7 @@ - + + + + @@ -173,6 +177,19 @@ + + + + When set to "true", the response is converted to the Flux which is sent as a reply + message payload. + The downstream flow might have a splitter to walk through that Flux. + Default is "false". + + + + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml index 3a4b996022..62427132ad 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml @@ -20,22 +20,23 @@ + web-client="webClient"/> + url="http://localhost/test2" + http-method="PUT" + request-channel="requests" + reply-timeout="1234" + extract-request-payload="false" + expected-response-type="java.lang.String" + mapped-request-headers="requestHeader1, requestHeader2" + mapped-response-headers="responseHeader" + reply-channel="replies" + charset="UTF-8" + order="77" + auto-startup="false" + transfer-cookies="true" + reply-to-flux="true"> diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java index 6995a6abcf..3f4276fbca 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java @@ -82,6 +82,7 @@ public class WebFluxOutboundGatewayParserTests { assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); assertEquals(true, handlerAccessor.getPropertyValue("extractPayload")); assertEquals(false, handlerAccessor.getPropertyValue("transferCookies")); + assertEquals(false, handlerAccessor.getPropertyValue("replyToFlux")); } @Test @@ -123,6 +124,7 @@ public class WebFluxOutboundGatewayParserTests { assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2")); assertEquals("responseHeader", mappedResponseHeaders[0]); assertEquals(true, handlerAccessor.getPropertyValue("transferCookies")); + assertEquals(true, handlerAccessor.getPropertyValue("replyToFlux")); } } 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 9c899d05e9..df7807d86b 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 @@ -28,7 +28,10 @@ import java.util.List; import org.junit.Test; import org.reactivestreams.Subscriber; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferFactory; import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; import org.springframework.http.client.reactive.ClientHttpConnector; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; @@ -42,6 +45,7 @@ import org.springframework.test.web.reactive.server.HttpHandlerConnector; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClientResponseException; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -54,7 +58,7 @@ import reactor.test.StepVerifier; public class WebFluxRequestExecutingMessageHandlerTests { @Test - public void testReactiveReturn() throws Throwable { + public void testReactiveReturn() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.OK); @@ -84,7 +88,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { } @Test - public void testReactiveErrorOneWay() throws Throwable { + public void testReactiveErrorOneWay() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.UNAUTHORIZED); @@ -114,7 +118,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { } @Test - public void testReactiveConnectErrorOneWay() throws Throwable { + public void testReactiveConnectErrorOneWay() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { throw new RuntimeException("Intentional connection error"); @@ -182,4 +186,50 @@ public class WebFluxRequestExecutingMessageHandlerTests { assertNull(replyMessage); } + @Test + @SuppressWarnings("unchecked") + public void testFluxReply() { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + response.setStatusCode(HttpStatus.OK); + response.getHeaders().setContentType(MediaType.TEXT_PLAIN); + + DataBufferFactory bufferFactory = response.bufferFactory(); + + Flux data = + Flux.just(bufferFactory.wrap("foo".getBytes()), + bufferFactory.wrap("bar".getBytes()), + bufferFactory.wrap("baz".getBytes())); + + return response.writeWith(data) + .then(Mono.defer(response::setComplete)); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + String destinationUri = "http://www.springsource.org/spring-integration"; + WebFluxRequestExecutingMessageHandler reactiveHandler = + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); + + QueueChannel replyChannel = new QueueChannel(); + reactiveHandler.setOutputChannel(replyChannel); + reactiveHandler.setExpectedResponseType(String.class); + reactiveHandler.setReplyToFlux(true); + + reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); + + Message receive = replyChannel.receive(10_000); + + assertNotNull(receive); + + assertThat(receive.getPayload(), instanceOf(Flux.class)); + + Flux flux = (Flux) receive.getPayload(); + + StepVerifier.create(flux) + .expectNext("foo", "bar", "baz") + .verifyComplete(); + } + } diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc index 2ee3a055e2..d85027441c 100644 --- a/src/reference/asciidoc/webflux.adoc +++ b/src/reference/asciidoc/webflux.adoc @@ -95,10 +95,13 @@ You can configure a `WebClient` instance to use: ---- -The `WebClient` `exchange()` operation returns a `Mono` which is mapped to the `AbstractIntegrationMessageBuilder` reactive support (using `Mono.map()`) as the output from the `WebFluxRequestExecutingMessageHandler`. +The `WebClient` `exchange()` operation returns a `Mono` which is mapped (using several `Mono.map()` steps) to an `AbstractIntegrationMessageBuilder` as the output from the `WebFluxRequestExecutingMessageHandler`. Together with the `ReactiveChannel` as an `outputChannel`, the `Mono` evaluation is deferred until a downstream subscription is made. Otherwise, it is treated as an `async` mode and the `Mono` response is adapted to an `SettableListenableFuture` for an asynchronous reply from the `WebFluxRequestExecutingMessageHandler`. - +The target payload of the output message depends on the `WebFluxRequestExecutingMessageHandler` configuration. +The `setExpectedResponseType(Class)` or `setExpectedResponseTypeExpression(Expression)` identifies the target type of the response body element conversion. +If `replyToFlux` is set to `true`, the response body is converted to a `Flux` with the provided `expectedResponseType` for each element and this `Flux` is sent as the payload downstream. +A <> afterwards can be used to iterate over this `Flux` in a reactive manner. Also see <> for more possible configuration options. @@ -237,4 +240,4 @@ public IntegrationFlow outboundReactive() { === WebFlux Header Mappings Since WebFlux components are fully based on the HTTP protocol there is no difference in the HTTP headers mapping. -See <> for more possible options and components to use for mapping headers. \ No newline at end of file +See <> for more possible options and components to use for mapping headers.