From b5ec98f025722554f820829a1a0914fdf546a678 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 20 Dec 2017 11:11:09 -0500 Subject: [PATCH] WebFlux improvements * add `BodyExtractor` support for Outbound part * add `ClientHttpResponseBodyExtractor` as identity function * add XML configuration for the Inbound part * document `BodyExtractor` and Inbound XML support Rename `replyToFlux` property to the `replyPayloadToFlux` Doc Polishing --- .../config/WebFluxInboundEndpointParser.java | 52 +++ .../config/WebFluxNamespaceHandler.java | 4 +- .../config/WebFluxOutboundGatewayParser.java | 5 +- .../dsl/WebFluxMessageHandlerSpec.java | 31 +- .../inbound/WebFluxInboundEndpoint.java | 4 +- ...WebFluxRequestExecutingMessageHandler.java | 36 +- .../ClientHttpResponseBodyExtractor.java | 37 ++ .../config/spring-integration-webflux-5.0.xsd | 325 +++++++++++++++++- ...boundChannelAdapterParserTests-context.xml | 48 +++ ...bFluxInboundChannelAdapterParserTests.java | 134 ++++++++ ...bFluxInboundGatewayParserTests-context.xml | 48 +++ .../WebFluxInboundGatewayParserTests.java | 134 ++++++++ ...FluxOutboundGatewayParserTests-context.xml | 6 +- .../WebFluxOutboundGatewayParserTests.java | 11 +- ...uxRequestExecutingMessageHandlerTests.java | 56 ++- src/reference/asciidoc/webflux.adoc | 48 ++- 16 files changed, 960 insertions(+), 19 deletions(-) create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxInboundEndpointParser.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/ClientHttpResponseBodyExtractor.java create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests-context.xml create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests-context.xml create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxInboundEndpointParser.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxInboundEndpointParser.java new file mode 100644 index 0000000000..6c542ced56 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxInboundEndpointParser.java @@ -0,0 +1,52 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.webflux.config; + +import org.w3c.dom.Element; + +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.HttpInboundEndpointParser; +import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint; + +/** + * @author Artem Bilan + * + * @since 5.0.1 + */ +public class WebFluxInboundEndpointParser extends HttpInboundEndpointParser { + + public WebFluxInboundEndpointParser(boolean expectReply) { + super(expectReply); + } + + @Override + protected Class getBeanClass(Element element) { + return WebFluxInboundEndpoint.class; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + super.doParse(element, parserContext, builder); + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "codec-configurer"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "requested-content-type-resolver"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reactive-adapter-registry"); + } + +} diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java index a97775b646..d47f6b9d3b 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2017-2018 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. @@ -28,6 +28,8 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa public class WebFluxNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { + registerBeanDefinitionParser("inbound-channel-adapter", new WebFluxInboundEndpointParser(false)); + registerBeanDefinitionParser("inbound-gateway", new WebFluxInboundEndpointParser(true)); registerBeanDefinitionParser("outbound-channel-adapter", new WebFluxOutboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-gateway", new WebFluxOutboundGatewayParser()); } 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 ea8ec03797..ff6164953a 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2017-2018 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. @@ -47,7 +47,8 @@ public class WebFluxOutboundGatewayParser extends HttpOutboundGatewayParser { .addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef)); } - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-to-flux"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-payload-to-flux"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "body-extractor"); return builder; } 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 65aabf3d69..8a6755bc3b 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 the original author or authors. + * Copyright 2017-2018 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. @@ -20,11 +20,16 @@ import java.net.URI; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; +import org.springframework.http.client.reactive.ClientHttpResponse; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.http.dsl.BaseHttpMessageHandlerSpec; import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler; +import org.springframework.web.reactive.function.BodyExtractor; import org.springframework.web.reactive.function.client.WebClient; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + /** * The {@link BaseHttpMessageHandlerSpec} implementation for the {@link WebFluxRequestExecutingMessageHandler}. * @@ -53,6 +58,30 @@ public class WebFluxMessageHandlerSpec this.webClient = webClient; } + /** + * 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 replyPayloadToFlux represent reply payload as a {@link Flux} or as a value from the {@link Mono}. + * @since 5.0.1 + * @see WebFluxRequestExecutingMessageHandler#setReplyPayloadToFlux(boolean) + */ + public void replyPayloadToFlux(boolean replyPayloadToFlux) { + this.target.setReplyPayloadToFlux(replyPayloadToFlux); + } + + /** + * Specify a {@link BodyExtractor} as an alternative to the {@code expectedResponseType} + * to allow to get low-level access to the received {@link ClientHttpResponse}. + * @param bodyExtractor the {@link BodyExtractor} to use. + * @since 5.0.1 + * @see WebFluxRequestExecutingMessageHandler#setBodyExtractor(BodyExtractor) + */ + public void bodyExtractor(BodyExtractor bodyExtractor) { + this.target.setBodyExtractor(bodyExtractor); + } + @Override protected boolean isClientSet() { return this.webClient != null; diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index e06967af93..010d7efd13 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -126,7 +126,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W @Override public String getComponentType() { - return super.getComponentType().replaceFirst("(http:)", "$1webflux-"); + return super.getComponentType().replaceFirst("http", "webflux"); } @Override 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 29e800000c..55ed5f3699 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 the original author or authors. + * Copyright 2017-2018 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. @@ -31,6 +31,7 @@ import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.ReactiveHttpInputMessage; import org.springframework.http.ResponseEntity; +import org.springframework.http.client.reactive.ClientHttpResponse; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.http.outbound.AbstractHttpRequestExecutingMessageHandler; import org.springframework.messaging.Message; @@ -62,7 +63,9 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx private final WebClient webClient; - private boolean replyToFlux; + private boolean replyPayloadToFlux; + + private BodyExtractor bodyExtractor; /** * Create a handler that will send requests to the provided URI. @@ -120,13 +123,25 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx * 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}. + * @param replyPayloadToFlux 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; + public void setReplyPayloadToFlux(boolean replyPayloadToFlux) { + this.replyPayloadToFlux = replyPayloadToFlux; + } + + /** + * Specify a {@link BodyExtractor} as an alternative to the {@code expectedResponseType} + * to allow to get low-level access to the received {@link ClientHttpResponse}. + * @param bodyExtractor the {@link BodyExtractor} to use. + * @since 5.0.1 + * @see #setExpectedResponseType(Class) + * @see #setExpectedResponseTypeExpression(Expression) + */ + public void setBodyExtractor(BodyExtractor bodyExtractor) { + this.bodyExtractor = bodyExtractor; } @Override @@ -191,7 +206,7 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx Mono bodyMono; if (expectedResponseType != null) { - if (this.replyToFlux) { + if (this.replyPayloadToFlux) { BodyExtractor, ReactiveHttpInputMessage> extractor; if (expectedResponseType instanceof ParameterizedTypeReference) { extractor = BodyExtractors.toFlux( @@ -215,6 +230,15 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx 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(); } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/ClientHttpResponseBodyExtractor.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/ClientHttpResponseBodyExtractor.java new file mode 100644 index 0000000000..3d7c8a4762 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/ClientHttpResponseBodyExtractor.java @@ -0,0 +1,37 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.webflux.support; + +import org.springframework.http.client.reactive.ClientHttpResponse; +import org.springframework.web.reactive.function.BodyExtractor; + +/** + * The {@link BodyExtractor} identity function implementation + * which just returns the provided {@link ClientHttpResponse}. + * + * @author Artem Bilan + * + * @since 5.0.1 + */ +public class ClientHttpResponseBodyExtractor implements BodyExtractor { + + @Override + public ClientHttpResponse extract(ClientHttpResponse response, Context context) { + return response; + } + +} 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 b9f8660cfe..6683204459 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 @@ -20,6 +20,315 @@ ]]> + + + + + Configures a Message Producing Endpoint for the + 'org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint' that + receives HTTP requests reactive manner and doesn't produce responses. + + + + + + + + Defines configuration for org.springframework.integration.http.inbound.RequestMapping + as RESTFul attributes for Spring Integration HTTP Inbound Endpoints. + + + + + + + Marks this endpoint as permitting cross origin requests (CORS). + + + + + + + Specifies a Message header as a result of expression evaluation + against ServletRequest and URI Variables. + + + + + + + + + Maximum amount of time in milliseconds to wait when sending + a message to the channel if such channel may block. + For example, a Queue Channel can block until space + is available if its maximum capacity has been reached. + + + + + + + A SpEL expression that resolves to an 'HttpStatus' code when rendering a response. + The expression must return an object which can be converted to a + 'org.springframework.http.HttpStatus' enum value. + The 'evaluationContext' has a 'BeanResolver' but no variables, so the usage of this attribute + is somewhat limited. + An example might be to resolve, at runtime, some scoped Bean that returns an + 'HttpStatus' value, or use a literal expression e.g. "201". + By default 'status-code-expression' is null, meaning that the default '200 OK' response status + will be returned. + The 'http:inbound-gateway' resolves the 'status code' from the 'http_statusCode' header of the reply + Message. + + + + + + + + + + + Configures a Messaging Gateway Endpoint for the + 'org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint' that + receives HTTP requests and produces responses reactive manner. + + + + + + + + + + Defines configuration for org.springframework.integration.http.inbound.RequestMapping + as RESTFul attributes for Spring Integration HTTP Inbound Endpoints. + + + + + + + + + The producible media types of the mapped request, narrowing the primary mapping. + The format is a sequence of media types ("text/plain", "application/*), + with a request only mapped if the Accept matches one of these media types. + Expressions can be negated by using the "!" operator, as in "!text/plain", which matches + all requests with a Accept other than "text/plain". + + + + + + + + + + + Marks this endpoint as permitting cross origin requests (CORS). + + + + + + + Specifies a Message header as a result of expression evaluation + against ServletRequest and URI Variables. + + + + + + + + + + + + + The receiving Message Channel of this endpoint. + + + + + + + + Specify whether only the reply Message's payload should be passed in the response. + If this is set to 'false', the entire Message will be used to generate the response. + The default is 'true'. + + + + + + + + + + + + + + + + + + + + + + A SpEL expression that resolves to an 'HttpStatus' code when rendering a response after + a 'reply-timeout'. + The expression must return an object which can be converted to a + 'org.springframework.http.HttpStatus' enum value. + The 'evaluationContext' has a 'BeanResolver' but no variables, so the usage of this attribute + is somewhat limited. + An example might be to resolve, at runtime, some scoped Bean that returns an + 'HttpStatus' value, or use a literal expression e.g. "504". + By default 'status-code-expression' is null, meaning that the default + '500 Internal Server Error' response status will be returned after a timeout. + When a timeout is not encountered, + the 'http:inbound-gateway' resolves the 'status code' from the 'http_statusCode' header of the reply + Message. + + + + + + + + + + + + + Comma-separated URI paths (e.g., /orderId/{order}). + Ant-style path patterns are also supported (e.g. /myPath/*.do). + + + + + + + Comma-separated HTTP Method names. Determines which types of Request are + allowed with this Endpoint. + + + + + + + + + + Target type for payload that is the conversion result of the request. + + + + + + + Allows you to specify SpEL expression to construct a Message payload + + + + + + + + + + + + Specifies a reference to org.springframework.integration.mapping.HeaderMapper + implementation bean. Only one of 'header-mapper' or 'mapped-request-headers' attributes + can be provided. + + + + + + + + + + + + + + + + + The MessagingGateway's 'error-channel' where to send an ErrorMessage in case + of Exception is caused from original message flow. + + + + + + + + + + + + A 'ServerCodecConfigurer' for the request readers and response writers. + By default the 'ServerCodecConfigurer#create()' factory is used. + + + + + + + + + + + + A 'RequestedContentTypeResolver' strategy to resolve the requested media types + for a 'ServerWebExchange'. + A 'HeaderContentTypeResolver' is used by default. + + + + + + + + + + + + A 'ReactiveAdapterRegistry' registry of adapters to adapt a + Reactive Streams 'Publisher' to/from. + + + + + @@ -177,7 +486,7 @@ - + When set to "true", the response is converted to the Flux which is sent as a reply @@ -190,6 +499,20 @@ + + + + + + + + + A reference to an org.springframework.web.reactive.function.BodyExtractor bean + which is used to to extract the content from the response. + Must be used instead of 'expected-response-type(-expression)' + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..e02bbe8b21 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests-context.xml @@ -0,0 +1,48 @@ + + + + + + + + + + +
+ + + + + + + + + + + + + + + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..578369fa18 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java @@ -0,0 +1,134 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.webflux.config; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.ResolvableType; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpression; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.integration.http.inbound.CrossOrigin; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.web.reactive.accept.RequestedContentTypeResolver; + +/** + * @author Artem Bilan + * + * @since 5.0.1 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class WebFluxInboundChannelAdapterParserTests { + + @Autowired + @Qualifier("reactiveMinimalConfig") + private WebFluxInboundEndpoint reactiveMinimalConfig; + + @Autowired + @Qualifier("reactiveFullConfig") + private WebFluxInboundEndpoint reactiveFullConfig; + + @Autowired + private MessageChannel requests; + + @Autowired + private HeaderMapper headerMapper; + + @Autowired + private ServerCodecConfigurer serverCodecConfigurer; + + @Autowired + private RequestedContentTypeResolver requestedContentTypeResolver; + + @Autowired + private ReactiveAdapterRegistry reactiveAdapterRegistry; + + @Test + public void reactiveMinimalConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveMinimalConfig); + assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel")); + assertTrue((boolean) endpointAccessor.getPropertyValue("autoStartup")); + assertFalse((boolean) endpointAccessor.getPropertyValue("expectReply")); + assertNull(endpointAccessor.getPropertyValue("statusCodeExpression")); + assertNull(endpointAccessor.getPropertyValue("payloadExpression")); + assertNull(endpointAccessor.getPropertyValue("headerExpressions")); + assertNull(endpointAccessor.getPropertyValue("crossOrigin")); + assertNull(endpointAccessor.getPropertyValue("requestPayloadType")); + + assertNotSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper")); + assertNotSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer")); + assertNotSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver")); + assertNotSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry")); + } + + @Test + @SuppressWarnings("unchecked") + public void reactiveFullConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfig); + assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel")); + assertNotNull(endpointAccessor.getPropertyValue("errorChannel")); + assertFalse((boolean) endpointAccessor.getPropertyValue("autoStartup")); + assertEquals(101, endpointAccessor.getPropertyValue("phase")); + assertFalse((boolean) endpointAccessor.getPropertyValue("expectReply")); + + assertEquals("'202'", + ((SpelExpression) endpointAccessor.getPropertyValue("statusCodeExpression")).getExpressionString()); + + assertEquals("payload", + ((SpelExpression) endpointAccessor.getPropertyValue("payloadExpression")).getExpressionString()); + + Map headerExpressions = + (Map) endpointAccessor.getPropertyValue("headerExpressions"); + + assertTrue(headerExpressions.containsKey("foo")); + + assertEquals("foo", headerExpressions.get("foo").getValue()); + + CrossOrigin crossOrigin = (CrossOrigin) endpointAccessor.getPropertyValue("crossOrigin"); + assertNotNull(crossOrigin); + assertArrayEquals(new String[] { "foo" }, crossOrigin.getOrigin()); + + assertEquals(ResolvableType.forClass(byte[].class), endpointAccessor.getPropertyValue("requestPayloadType")); + + assertSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper")); + assertSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer")); + assertSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver")); + assertSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry")); + } + +} diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests-context.xml new file mode 100644 index 0000000000..9f30c769e5 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests-context.xml @@ -0,0 +1,48 @@ + + + + + + + + + + +
+ + + + + + + + + + + + + + + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java new file mode 100644 index 0000000000..a64321a379 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java @@ -0,0 +1,134 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.webflux.config; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.ResolvableType; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpression; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.integration.http.inbound.CrossOrigin; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.web.reactive.accept.RequestedContentTypeResolver; + +/** + * @author Artem Bilan + * + * @since 5.0.1 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class WebFluxInboundGatewayParserTests { + + @Autowired + @Qualifier("reactiveMinimalConfig") + private WebFluxInboundEndpoint reactiveMinimalConfig; + + @Autowired + @Qualifier("reactiveFullConfig") + private WebFluxInboundEndpoint reactiveFullConfig; + + @Autowired + private MessageChannel requests; + + @Autowired + private HeaderMapper headerMapper; + + @Autowired + private ServerCodecConfigurer serverCodecConfigurer; + + @Autowired + private RequestedContentTypeResolver requestedContentTypeResolver; + + @Autowired + private ReactiveAdapterRegistry reactiveAdapterRegistry; + + @Test + public void reactiveMinimalConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveMinimalConfig); + assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel")); + assertTrue((boolean) endpointAccessor.getPropertyValue("autoStartup")); + assertTrue((boolean) endpointAccessor.getPropertyValue("expectReply")); + assertNull(endpointAccessor.getPropertyValue("statusCodeExpression")); + assertNull(endpointAccessor.getPropertyValue("payloadExpression")); + assertNull(endpointAccessor.getPropertyValue("headerExpressions")); + assertNull(endpointAccessor.getPropertyValue("crossOrigin")); + assertNull(endpointAccessor.getPropertyValue("requestPayloadType")); + + assertNotSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper")); + assertNotSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer")); + assertNotSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver")); + assertNotSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry")); + } + + @Test + @SuppressWarnings("unchecked") + public void reactiveFullConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfig); + assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel")); + assertNotNull(endpointAccessor.getPropertyValue("errorChannel")); + assertFalse((boolean) endpointAccessor.getPropertyValue("autoStartup")); + assertEquals(101, endpointAccessor.getPropertyValue("phase")); + assertTrue((boolean) endpointAccessor.getPropertyValue("expectReply")); + + assertEquals("'504'", + ((SpelExpression) endpointAccessor.getPropertyValue("statusCodeExpression")).getExpressionString()); + + assertEquals("payload", + ((SpelExpression) endpointAccessor.getPropertyValue("payloadExpression")).getExpressionString()); + + Map headerExpressions = + (Map) endpointAccessor.getPropertyValue("headerExpressions"); + + assertTrue(headerExpressions.containsKey("foo")); + + assertEquals("foo", headerExpressions.get("foo").getValue()); + + CrossOrigin crossOrigin = (CrossOrigin) endpointAccessor.getPropertyValue("crossOrigin"); + assertNotNull(crossOrigin); + assertArrayEquals(new String[] { "foo" }, crossOrigin.getOrigin()); + + assertEquals(ResolvableType.forClass(byte[].class), endpointAccessor.getPropertyValue("requestPayloadType")); + + assertSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper")); + assertSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer")); + assertSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver")); + assertSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry")); + } + +} 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 62427132ad..b95214542e 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 @@ -36,8 +36,12 @@ order="77" auto-startup="false" transfer-cookies="true" - reply-to-flux="true"> + reply-payload-to-flux="true" + body-extractor="bodyExtractor"> + + 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 3f4276fbca..7e8700215b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -40,6 +40,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.ObjectUtils; +import org.springframework.web.reactive.function.BodyExtractor; import org.springframework.web.reactive.function.client.WebClient; /** @@ -62,6 +63,9 @@ public class WebFluxOutboundGatewayParserTests { @Autowired private WebClient webClient; + @Autowired + private BodyExtractor bodyExtractor; + @Autowired private ApplicationContext applicationContext; @@ -82,7 +86,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")); + assertEquals(false, handlerAccessor.getPropertyValue("replyPayloadToFlux")); } @Test @@ -124,7 +128,8 @@ public class WebFluxOutboundGatewayParserTests { assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2")); assertEquals("responseHeader", mappedResponseHeaders[0]); assertEquals(true, handlerAccessor.getPropertyValue("transferCookies")); - assertEquals(true, handlerAccessor.getPropertyValue("replyToFlux")); + assertEquals(true, handlerAccessor.getPropertyValue("replyPayloadToFlux")); + assertSame(this.bodyExtractor, handlerAccessor.getPropertyValue("bodyExtractor")); } } 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 df7807d86b..6ee95fd9e6 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 @@ -18,6 +18,7 @@ package org.springframework.integration.webflux.outbound; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; @@ -33,11 +34,13 @@ 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.http.client.reactive.ClientHttpResponse; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.webflux.support.ClientHttpResponseBodyExtractor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.support.ErrorMessage; @@ -215,7 +218,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { QueueChannel replyChannel = new QueueChannel(); reactiveHandler.setOutputChannel(replyChannel); reactiveHandler.setExpectedResponseType(String.class); - reactiveHandler.setReplyToFlux(true); + reactiveHandler.setReplyPayloadToFlux(true); reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); @@ -232,4 +235,55 @@ public class WebFluxRequestExecutingMessageHandlerTests { .verifyComplete(); } + @Test + public void testClientHttpResponseAsReply() { + 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.setBodyExtractor(new ClientHttpResponseBodyExtractor()); + + reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); + + Message receive = replyChannel.receive(10_000); + + assertNotNull(receive); + + assertThat(receive.getPayload(), instanceOf(ClientHttpResponse.class)); + + ClientHttpResponse response = (ClientHttpResponse) receive.getPayload(); + + + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals(MediaType.TEXT_PLAIN, response.getHeaders().getContentType()); + + StepVerifier.create( + response.getBody() + .map(dataBuffer -> new String(dataBuffer.asByteBuffer().array()))) + .expectNext("foo", "bar", "baz") + .verifyComplete(); + } + + } diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc index d85027441c..4abf0d3d63 100644 --- a/src/reference/asciidoc/webflux.adoc +++ b/src/reference/asciidoc/webflux.adoc @@ -100,9 +100,13 @@ Together with the `ReactiveChannel` as an `outputChannel`, the `Mono)` 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. +If the `replyPayloadToFlux` 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. +In addition a `BodyExtractor` can be injected into the `WebFluxRequestExecutingMessageHandler` instead of `expectedResponseType` and `replyPayloadToFlux` properties. +It can be used for low-level access to the `ClientHttpResponse` and more control over body and HTTP headers conversion. +The `ClientHttpResponseBodyExtractor` is provided out-of-the-box as identity function to produce downstream the whole `ClientHttpResponse` and any other possible custom logic. + Also see <> for more possible configuration options. [[webflux-namespace]] @@ -133,6 +137,48 @@ To include it in your configuration, simply provide the following namespace decl ==== Inbound +To configure Spring Integration WebFlux via XML you may use appropriate components from the mentioned `int-webflux` namespace - `inbound-channel-adapter` or `inbound-gateway` according request/response requirements respectively: + +[source,xml] +---- + + + +
+ + + + + +
+ +---- ==== Outbound