From 84d60f4ab49892aac4f3a9195ea67ed8689746f5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 11 Aug 2017 10:37:08 -0400 Subject: [PATCH] INT-4315: Add WebFlux module JIRA: https://jira.spring.io/browse/INT-4315 Move Reactive components outside of HTTP module to the new WebFlux one, including XSD, tests and documentation Make an appropriate polishing for the `http.adoc` with cross-link to the `webflux.adoc` Exclude transitive `spring-webmvc` for the `spring-integration-webflux` --- build.gradle | 23 +- ...tpIntegrationConfigurationInitializer.java | 31 --- .../http/config/HttpNamespaceHandler.java | 2 - .../HttpOutboundChannelAdapterParser.java | 52 ++-- .../config/HttpOutboundGatewayParser.java | 51 ++-- .../http/dsl/BaseHttpMessageHandlerSpec.java | 2 +- .../integration/http/dsl/Http.java | 206 --------------- .../dsl/HttpInboundEndpointSupportSpec.java | 2 +- ...tegrationRequestMappingHandlerMapping.java | 2 +- .../http/support/HttpContextUtils.java | 15 -- .../config/spring-integration-http-5.0.xsd | 157 ----------- ...boundChannelAdapterParserTests-context.xml | 8 - ...HttpOutboundChannelAdapterParserTests.java | 36 --- ...HttpOutboundGatewayParserTests-context.xml | 23 -- .../HttpOutboundGatewayParserTests.java | 73 ------ .../integration/http/dsl/HttpDslTests.java | 130 ---------- .../config/spring-integration-syslog-5.0.xsd | 2 +- ...uxIntegrationConfigurationInitializer.java | 83 ++++++ .../config/WebFluxNamespaceHandler.java | 35 +++ .../WebFluxOutboundChannelAdapterParser.java | 52 ++++ .../config/WebFluxOutboundGatewayParser.java | 52 ++++ .../webflux/config/package-info.java | 4 + .../integration/webflux/dsl/WebFlux.java | 245 ++++++++++++++++++ .../dsl/WebFluxInboundEndpointSpec.java | 19 +- .../dsl/WebFluxMessageHandlerSpec.java | 26 +- .../integration/webflux/dsl/package-info.java | 4 + .../IntegrationHandlerResultHandler.java | 8 +- .../inbound/WebFluxInboundEndpoint.java | 11 +- ...tegrationRequestMappingHandlerMapping.java | 49 ++-- .../webflux/inbound/package-info.java | 4 + ...WebFluxRequestExecutingMessageHandler.java | 19 +- .../webflux/outbound/package-info.java | 4 + .../webflux/support/WebFluxContextUtils.java | 50 ++++ .../webflux/support/package-info.java | 4 + .../main/resources/META-INF/spring.factories | 2 + .../main/resources/META-INF/spring.handlers | 1 + .../main/resources/META-INF/spring.schemas | 2 + .../main/resources/META-INF/spring.tooling | 4 + .../config/spring-integration-webflux-5.0.xsd | 181 +++++++++++++ .../config/spring-integration-webflux.gif | Bin 0 -> 578 bytes ...boundChannelAdapterParserTests-context.xml | 21 ++ ...HttpOutboundChannelAdapterParserTests.java | 88 +++++++ ...HttpOutboundGatewayParserTests-context.xml | 42 +++ .../HttpOutboundGatewayParserTests.java | 128 +++++++++ .../webflux/dsl/WebFluxDslTests.java | 237 +++++++++++++++++ .../inbound/WebFluxInboundEndpointTests.java | 15 +- ...uxRequestExecutingMessageHandlerTests.java | 23 +- .../src/test/resources/log4j.properties | 9 + src/reference/asciidoc/http.adoc | 175 +------------ src/reference/asciidoc/index.adoc | 2 + src/reference/asciidoc/webflux.adoc | 240 +++++++++++++++++ src/reference/asciidoc/whats-new.adoc | 6 +- 52 files changed, 1657 insertions(+), 1003 deletions(-) create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxIntegrationConfigurationInitializer.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParser.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/package-info.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFlux.java rename spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java => spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java (63%) rename spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java => spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java (59%) create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/package-info.java rename {spring-integration-http/src/main/java/org/springframework/integration/http => spring-integration-webflux/src/main/java/org/springframework/integration/webflux}/inbound/IntegrationHandlerResultHandler.java (85%) rename spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java => spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java (97%) rename spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java => spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java (76%) create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/package-info.java rename spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java => spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java (87%) create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/package-info.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/WebFluxContextUtils.java create mode 100644 spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/package-info.java create mode 100644 spring-integration-webflux/src/main/resources/META-INF/spring.factories create mode 100644 spring-integration-webflux/src/main/resources/META-INF/spring.handlers create mode 100644 spring-integration-webflux/src/main/resources/META-INF/spring.schemas create mode 100644 spring-integration-webflux/src/main/resources/META-INF/spring.tooling create mode 100644 spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd create mode 100644 spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux.gif create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests-context.xml create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests.java create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests-context.xml create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests.java create mode 100644 spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java rename spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java => spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java (90%) rename spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java => spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java (87%) create mode 100644 spring-integration-webflux/src/test/resources/log4j.properties create mode 100644 src/reference/asciidoc/webflux.adoc diff --git a/build.gradle b/build.gradle index 584b6e1699..91bab61a22 100644 --- a/build.gradle +++ b/build.gradle @@ -396,17 +396,13 @@ project('spring-integration-http') { description = 'Spring Integration HTTP Support' dependencies { compile project(":spring-integration-core") - compile "org.springframework:spring-web:$springVersion" - compile ("org.springframework:spring-webmvc:$springVersion", optional) - compile ("org.springframework:spring-webflux:$springVersion", optional) + compile "org.springframework:spring-webmvc:$springVersion" compile ("javax.servlet:javax.servlet-api:$servletApiVersion", provided) compile ("com.rometools:rome:$romeToolsVersion", optional) - compile ("io.projectreactor.ipc:reactor-netty:$reactorNettyVersion" , optional) testCompile project(":spring-integration-security") testCompile "org.springframework.security:spring-security-config:$springSecurityVersion" testCompile "org.springframework.security:spring-security-test:$springSecurityVersion" - testCompile "io.projectreactor:reactor-test:$reactorVersion" } } @@ -642,7 +638,6 @@ project('spring-integration-twitter') { exclude group: 'org.springframework', module: 'spring-expression' exclude group: 'org.springframework', module: 'spring-web' exclude group: 'org.springframework', module: 'spring-webmvc' - } compile("javax.activation:activation:$javaxActivationVersion", optional) testCompile project(":spring-integration-redis") @@ -651,6 +646,22 @@ project('spring-integration-twitter') { } } +project('spring-integration-webflux') { + description = 'Spring Integration HTTP Support' + dependencies { + compile (project(":spring-integration-http")) { + exclude group: 'org.springframework', module: 'spring-webmvc' + } + compile "org.springframework:spring-webflux:$springVersion" + compile ("io.projectreactor.ipc:reactor-netty:$reactorNettyVersion" , optional) + + testCompile "org.springframework:spring-webmvc:$springVersion" + testCompile "org.springframework.security:spring-security-config:$springSecurityVersion" + testCompile "org.springframework.security:spring-security-test:$springSecurityVersion" + testCompile "io.projectreactor:reactor-test:$reactorVersion" + } +} + project('spring-integration-websocket') { description = 'Spring Integration WebSockets Support' dependencies { diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java index 82c9e55178..749511fc95 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java @@ -23,14 +23,10 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.integration.config.IntegrationConfigurationInitializer; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.springframework.integration.http.inbound.IntegrationHandlerResultHandler; import org.springframework.integration.http.inbound.IntegrationRequestMappingHandlerMapping; -import org.springframework.integration.http.inbound.ReactiveIntegrationRequestMappingHandlerMapping; import org.springframework.integration.http.support.HttpContextUtils; /** @@ -48,7 +44,6 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException { if (beanFactory instanceof BeanDefinitionRegistry) { registerRequestMappingHandlerMappingIfNecessary((BeanDefinitionRegistry) beanFactory); - registerReactiveRequestMappingHandlerMappingIfNecessary((BeanDefinitionRegistry) beanFactory); } else { logger.warn("'IntegrationRequestMappingHandlerMapping' isn't registered because 'beanFactory'" + @@ -78,30 +73,4 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi } } - /** - * Registers a {@link ReactiveIntegrationRequestMappingHandlerMapping} - * which could also be overridden by the user by simply registering - * a {@link ReactiveIntegrationRequestMappingHandlerMapping} {@code } with 'id' - * {@link HttpContextUtils#REACTIVE_HANDLER_MAPPING_BEAN_NAME}. - *

- * In addition, checks if the {@code org.springframework.web.reactive.result.method.RequestMappingInfo} - * class is present on the classpath. - * When Spring Integration HTTP is used only as an HTTP client, there is no reason to use and register - * the HTTP server components. - */ - private void registerReactiveRequestMappingHandlerMappingIfNecessary(BeanDefinitionRegistry registry) { - if (HttpContextUtils.WEB_FLUX_PRESENT && - !registry.containsBeanDefinition(HttpContextUtils.REACTIVE_HANDLER_MAPPING_BEAN_NAME)) { - BeanDefinitionBuilder requestMappingBuilder = - BeanDefinitionBuilder.genericBeanDefinition(ReactiveIntegrationRequestMappingHandlerMapping.class); - requestMappingBuilder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); - requestMappingBuilder.addPropertyValue(IntegrationNamespaceUtils.ORDER, 0); - registry.registerBeanDefinition(HttpContextUtils.REACTIVE_HANDLER_MAPPING_BEAN_NAME, - requestMappingBuilder.getBeanDefinition()); - - BeanDefinitionReaderUtils.registerWithGeneratedName( - new RootBeanDefinition(IntegrationHandlerResultHandler.class), registry); - } - } - } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpNamespaceHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpNamespaceHandler.java index 93ceb5d3f4..74f10dd9af 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpNamespaceHandler.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpNamespaceHandler.java @@ -34,8 +34,6 @@ public class HttpNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("inbound-gateway", new HttpInboundEndpointParser(true)); registerBeanDefinitionParser("outbound-channel-adapter", new HttpOutboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-gateway", new HttpOutboundGatewayParser()); - registerBeanDefinitionParser("outbound-reactive-channel-adapter", new HttpOutboundChannelAdapterParser()); - registerBeanDefinitionParser("outbound-reactive-gateway", new HttpOutboundGatewayParser()); registerBeanDefinitionParser("graph-controller", new IntegrationGraphControllerParser()); } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParser.java index 7dfad2ec0c..63a2f30a57 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParser.java @@ -18,13 +18,13 @@ package org.springframework.integration.http.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; -import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; import org.springframework.util.StringUtils; /** @@ -42,41 +42,13 @@ public class HttpOutboundChannelAdapterParser extends AbstractOutboundChannelAda @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder; - boolean reactive = element.getLocalName().contains("reactive"); - if (reactive) { - builder = BeanDefinitionBuilder.genericBeanDefinition(ReactiveHttpRequestExecutingMessageHandler.class); - } - else { - builder = BeanDefinitionBuilder.genericBeanDefinition(HttpRequestExecutingMessageHandler.class); - } + BeanDefinitionBuilder builder = getBuilder(element, parserContext); builder.addPropertyValue("expectReply", false); HttpAdapterParsingUtils.configureUrlConstructorArg(element, parserContext, builder); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encode-uri"); HttpAdapterParsingUtils.setHttpMethodOrExpression(element, parserContext, builder); - if (reactive) { - String webClientRef = element.getAttribute("web-client"); - - if (StringUtils.hasText(webClientRef)) { - builder.addConstructorArgReference(webClientRef); - } - } - else { - String restTemplateRef = element.getAttribute("rest-template"); - - if (StringUtils.hasText(restTemplateRef)) { - HttpAdapterParsingUtils.verifyNoRestTemplateAttributes(element, parserContext); - builder.addConstructorArgReference(restTemplateRef); - } - else { - for (String referenceAttributeName : HttpAdapterParsingUtils.SYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, referenceAttributeName); - } - } - } - String headerMapper = element.getAttribute("header-mapper"); String mappedRequestHeaders = element.getAttribute("mapped-request-headers"); if (StringUtils.hasText(headerMapper)) { @@ -101,4 +73,24 @@ public class HttpOutboundChannelAdapterParser extends AbstractOutboundChannelAda return builder.getBeanDefinition(); } + protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(HttpRequestExecutingMessageHandler.class); + + String restTemplateRef = element.getAttribute("rest-template"); + + if (StringUtils.hasText(restTemplateRef)) { + HttpAdapterParsingUtils.verifyNoRestTemplateAttributes(element, parserContext); + builder.getBeanDefinition() + .getConstructorArgumentValues() + .addIndexedArgumentValue(1, new RuntimeBeanReference(restTemplateRef)); + } + else { + for (String referenceAttributeName : HttpAdapterParsingUtils.SYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, referenceAttributeName); + } + } + return builder; + } + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundGatewayParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundGatewayParser.java index f23e4c1148..3548bcd3e6 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundGatewayParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpOutboundGatewayParser.java @@ -18,12 +18,12 @@ package org.springframework.integration.http.config; 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.AbstractConsumerEndpointParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; -import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; import org.springframework.util.StringUtils; /** @@ -44,39 +44,12 @@ public class HttpOutboundGatewayParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder; - boolean reactive = element.getLocalName().contains("reactive"); - if (reactive) { - builder = BeanDefinitionBuilder.genericBeanDefinition(ReactiveHttpRequestExecutingMessageHandler.class); - } - else { - builder = BeanDefinitionBuilder.genericBeanDefinition(HttpRequestExecutingMessageHandler.class); - } + BeanDefinitionBuilder builder = getBuilder(element, parserContext); HttpAdapterParsingUtils.configureUrlConstructorArg(element, parserContext, builder); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encode-uri"); HttpAdapterParsingUtils.setHttpMethodOrExpression(element, parserContext, builder); - if (reactive) { - String webClientRef = element.getAttribute("web-client"); - if (StringUtils.hasText(webClientRef)) { - builder.addConstructorArgReference(webClientRef); - } - } - else { - String restTemplateRef = element.getAttribute("rest-template"); - - if (StringUtils.hasText(restTemplateRef)) { - HttpAdapterParsingUtils.verifyNoRestTemplateAttributes(element, parserContext); - builder.addConstructorArgReference(restTemplateRef); - } - else { - for (String referenceAttributeName : HttpAdapterParsingUtils.SYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, referenceAttributeName); - } - } - } - String headerMapper = element.getAttribute("header-mapper"); String mappedRequestHeaders = element.getAttribute("mapped-request-headers"); String mappedResponseHeaders = element.getAttribute("mapped-response-headers"); @@ -113,4 +86,24 @@ public class HttpOutboundGatewayParser extends AbstractConsumerEndpointParser { return builder; } + protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(HttpRequestExecutingMessageHandler.class); + + String restTemplateRef = element.getAttribute("rest-template"); + + if (StringUtils.hasText(restTemplateRef)) { + HttpAdapterParsingUtils.verifyNoRestTemplateAttributes(element, parserContext); + builder.getBeanDefinition() + .getConstructorArgumentValues() + .addIndexedArgumentValue(1, new RuntimeBeanReference(restTemplateRef)); + } + else { + for (String referenceAttributeName : HttpAdapterParsingUtils.SYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, referenceAttributeName); + } + } + return builder; + } + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpMessageHandlerSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpMessageHandlerSpec.java index ee53735bc9..d4354e2662 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpMessageHandlerSpec.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpMessageHandlerSpec.java @@ -61,7 +61,7 @@ public abstract class BaseHttpMessageHandlerSpec the expected payload type. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static

ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(Function, ?> uriFunction) { - return outboundReactiveChannelAdapter(new FunctionExpression<>(uriFunction)); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter - * based on provided SpEL {@link Expression} to evaluate target {@code uri} - * against request message. - * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(Expression uriExpression) { - return outboundReactiveChannelAdapter(uriExpression, null); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter - * based on provided {@link URI} and {@link WebClient}. - * @param uri the {@link URI} to send requests. - * @param webClient {@link WebClient} to use. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(URI uri, WebClient webClient) { - return new ReactiveHttpMessageHandlerSpec(uri, webClient) - .expectReply(false); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter - * based on provided {@code uri} and {@link WebClient}. - * @param uri the {@code uri} to send requests. - * @param webClient {@link WebClient} to use. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(String uri, WebClient webClient) { - return new ReactiveHttpMessageHandlerSpec(uri, webClient) - .expectReply(false); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter - * based on provided {@code Function} to evaluate target {@code uri} against request message - * and {@link WebClient} for HTTP exchanges. - * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. - * @param webClient {@link WebClient} to use. - * @param

the expected payload type. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static

ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(Function, ?> uriFunction, - WebClient webClient) { - return outboundReactiveChannelAdapter(new FunctionExpression<>(uriFunction), webClient); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter - * based on provided SpEL {@link Expression} to evaluate target {@code uri} - * against request message and {@link WebClient} for HTTP exchanges. - * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. - * @param webClient {@link WebClient} to use. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(Expression uriExpression, - WebClient webClient) { - return new ReactiveHttpMessageHandlerSpec(uriExpression, webClient) - .expectReply(false); - } - /** * Create an {@link HttpMessageHandlerSpec} builder for request-reply gateway based on provided {@link URI}. * @param uri the {@link URI} to send requests. @@ -307,96 +213,6 @@ public final class Http { return new HttpMessageHandlerSpec(uriExpression, restTemplate); } - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided {@link URI}. - * @param uri the {@link URI} to send requests. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(URI uri) { - return outboundReactiveGateway(uri, null); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided {@code uri}. - * @param uri the {@code uri} to send requests. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(String uri) { - return outboundReactiveGateway(uri, null); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided {@code Function} to evaluate target {@code uri} against request message. - * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. - * @param

the expected payload type. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static

ReactiveHttpMessageHandlerSpec outboundReactiveGateway(Function, ?> uriFunction) { - return outboundReactiveGateway(new FunctionExpression<>(uriFunction)); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided SpEL {@link Expression} to evaluate target {@code uri} against request message. - * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(Expression uriExpression) { - return outboundReactiveGateway(uriExpression, null); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided {@link URI} and {@link WebClient}. - * @param uri the {@link URI} to send requests. - * @param webClient {@link WebClient} to use. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(URI uri, WebClient webClient) { - return new ReactiveHttpMessageHandlerSpec(uri, webClient); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided {@code uri} and {@link WebClient}. - * @param uri the {@code uri} to send requests. - * @param webClient {@link WebClient} to use. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(String uri, WebClient webClient) { - return new ReactiveHttpMessageHandlerSpec(uri, webClient); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided {@code Function} to evaluate target {@code uri} against request message - * and {@link WebClient} for HTTP exchanges. - * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. - * @param webClient {@link WebClient} to use. - * @param

the expected payload type. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static

ReactiveHttpMessageHandlerSpec outboundReactiveGateway(Function, ?> uriFunction, - WebClient webClient) { - return outboundReactiveGateway(new FunctionExpression<>(uriFunction), webClient); - } - - /** - * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway - * based on provided SpEL {@link Expression} to evaluate target {@code uri} - * against request message and {@link WebClient} for HTTP exchanges. - * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. - * @param webClient {@link WebClient} to use. - * @return the ReactiveHttpMessageHandlerSpec instance - */ - public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(Expression uriExpression, - WebClient webClient) { - return new ReactiveHttpMessageHandlerSpec(uriExpression, webClient); - } - /** * Create an {@link HttpControllerEndpointSpec} builder for one-way adapter * based on the provided MVC {@code viewName} and {@code path} array for mapping. @@ -468,28 +284,6 @@ public final class Http { return new HttpRequestHandlerEndpointSpec(new HttpRequestHandlingMessagingGateway(), path); } - - /** - * Create an {@link ReactiveHttpInboundEndpointSpec} builder for one-way reactive adapter - * based on the provided {@code path} array for mapping. - * @param path the path mapping URIs (e.g. "/myPath.do"). - * @return the ReactiveHttpInboundEndpointSpec instance - */ - public static ReactiveHttpInboundEndpointSpec inboundReactiveChannelAdapter(String... path) { - ReactiveHttpInboundEndpoint httpInboundChannelAdapter = new ReactiveHttpInboundEndpoint(false); - return new ReactiveHttpInboundEndpointSpec(httpInboundChannelAdapter, path); - } - - /** - * Create an {@link ReactiveHttpInboundEndpointSpec} builder for request-reply reactive gateway - * based on the provided {@code path} array for mapping. - * @param path the path mapping URIs (e.g. "/myPath.do"). - * @return the ReactiveHttpInboundEndpointSpec instance - */ - public static ReactiveHttpInboundEndpointSpec inboundReactiveGateway(String... path) { - return new ReactiveHttpInboundEndpointSpec(new ReactiveHttpInboundEndpoint(), path); - } - private Http() { super(); } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java index 682e213cc1..cba75f78fb 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java @@ -57,7 +57,7 @@ public abstract class HttpInboundEndpointSupportSpec explicitHeaderMapper; - HttpInboundEndpointSupportSpec(E gateway, String... path) { + protected HttpInboundEndpointSupportSpec(E gateway, String... path) { super(gateway); this.requestMapping.setPathPatterns(path); this.target.setRequestMapping(this.requestMapping); diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java index 6d99fb718c..a64d64f517 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java @@ -50,7 +50,7 @@ import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandl * elements, if there is none registered, yet. However it can be configured as a regular * bean with appropriate configuration for {@link RequestMappingHandlerMapping}. * It is recommended to have only one similar bean in the application context using the 'id' - * {@link org.springframework.integration.http.support.HttpContextUtils#REACTIVE_HANDLER_MAPPING_BEAN_NAME}. + * {@link org.springframework.integration.http.support.HttpContextUtils#HANDLER_MAPPING_BEAN_NAME}. *

* In most cases, Spring MVC offers to configure Request Mapping via * {@code org.springframework.stereotype.Controller} and diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java b/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java index 00b8863b6b..9e98ae0a4c 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java @@ -48,27 +48,12 @@ public final class HttpContextUtils { ClassUtils.isPresent("org.springframework.web.servlet.DispatcherServlet", HttpContextUtils.class.getClassLoader()); - /** - * The {@code boolean} flag to indicate if the - * {@code org.springframework.web.reactive.result.method.RequestMappingInfo} - * is present in the CLASSPATH to allow to register the Integration server reactive components. - */ - public static final boolean WEB_FLUX_PRESENT = - ClassUtils.isPresent("org.springframework.web.reactive.result.method.RequestMappingInfo", - HttpContextUtils.class.getClassLoader()); - /** * The name for the infrastructure * {@link org.springframework.integration.http.inbound.IntegrationRequestMappingHandlerMapping} bean. */ public static final String HANDLER_MAPPING_BEAN_NAME = "integrationRequestMappingHandlerMapping"; - /** - * The name for the infrastructure - * {@link org.springframework.integration.http.inbound.ReactiveIntegrationRequestMappingHandlerMapping} bean. - */ - public static final String REACTIVE_HANDLER_MAPPING_BEAN_NAME = "reactiveIntegrationRequestMappingHandlerMapping"; - /** * Represents the environment property for the {@code IntegrationGraphController} request mapping path. */ diff --git a/spring-integration-http/src/main/resources/org/springframework/integration/http/config/spring-integration-http-5.0.xsd b/spring-integration-http/src/main/resources/org/springframework/integration/http/config/spring-integration-http-5.0.xsd index a7c006b2a1..aaf741049c 100644 --- a/spring-integration-http/src/main/resources/org/springframework/integration/http/config/spring-integration-http-5.0.xsd +++ b/spring-integration-http/src/main/resources/org/springframework/integration/http/config/spring-integration-http-5.0.xsd @@ -401,55 +401,6 @@ - - - - Configures a Consumer Endpoint for the - 'org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler' - with 'expectReply = false' that sends HTTP requests based on incoming messages. - - - - - - - - Specify an expression for URI variable placeholder within 'url'. - This element is mutually exclusive with 'uri-variables-expression' attribute. - - - - - - - - - - - - - - - - - A reference to an org.springframework.web.reactive.function.client.WebClient bean - which is used to send to send the HTTP Requests reactive manner. - - - - - - - Specify whether the outbound message's payload should be extracted - when preparing the request body. Otherwise the Message instance itself - will be serialized. - The default value is 'true'. - - - - - - @@ -544,114 +495,6 @@ - - - - Configures a Consumer Endpoint for the - 'org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler' - that sends HTTP requests based on incoming messages and expects HTTP responses. - - - - - - - - - - Specify an expression for URI variable placeholder within 'url'. - - - - - - - - - - - - - - - - The receiving Message Channel of this endpoint. - - - - - - - - - - - - Specifies whether the outbound message's payload should be extracted - when preparing the request body. Otherwise the Message instance itself - will be serialized. - The default value is 'true'. - - - - - - - - - - - - - - - - - - - - - - - A reference to an org.springframework.web.reactive.function.client.WebClient bean - which is used to send to send the HTTP Requests reactive manner. - - - - - - - - diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests-context.xml b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests-context.xml index de86d0ee94..46afbc92d1 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests-context.xml @@ -17,16 +17,8 @@ - - - - - - - - @@ -43,26 +40,6 @@ - - - - - - diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java index 4a1ce768ce..f322259e2e 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java @@ -19,7 +19,6 @@ package org.springframework.integration.http.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -58,7 +57,6 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ObjectUtils; import org.springframework.web.client.ResponseErrorHandler; -import org.springframework.web.reactive.function.client.WebClient; /** * @author Mark Fisher @@ -79,14 +77,6 @@ public class HttpOutboundGatewayParserTests { @Qualifier("fullConfig") private AbstractEndpoint fullConfigEndpoint; - @Autowired - @Qualifier("reactiveMinimalConfig") - private AbstractEndpoint reactiveMinimalConfigEndpoint; - - @Autowired - @Qualifier("reactiveFullConfig") - private AbstractEndpoint reactiveFullConfigEndpoint; - @Autowired @Qualifier("withUrlExpression") private AbstractEndpoint withUrlExpressionEndpoint; @@ -99,9 +89,6 @@ public class HttpOutboundGatewayParserTests { @Qualifier("withPoller1") private AbstractEndpoint withPoller1; - @Autowired - private WebClient webClient; - @Autowired private ApplicationContext applicationContext; @@ -178,66 +165,6 @@ public class HttpOutboundGatewayParserTests { assertEquals(true, handlerAccessor.getPropertyValue("transferCookies")); } - @Test - public void reactiveMinimalConfig() { - Object handler = new DirectFieldAccessor(this.reactiveMinimalConfigEndpoint).getPropertyValue("handler"); - Object requestChannel = new DirectFieldAccessor(this.reactiveMinimalConfigEndpoint) - .getPropertyValue("inputChannel"); - assertEquals(this.applicationContext.getBean("requests"), requestChannel); - DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); - Object replyChannel = handlerAccessor.getPropertyValue("outputChannel"); - assertNull(replyChannel); - assertSame(this.webClient, handlerAccessor.getPropertyValue("webClient")); - Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression"); - assertEquals("http://localhost/test1", uriExpression.getValue()); - assertEquals(HttpMethod.POST.name(), - TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); - assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); - assertEquals(true, handlerAccessor.getPropertyValue("extractPayload")); - assertEquals(false, handlerAccessor.getPropertyValue("transferCookies")); - } - - @Test - @SuppressWarnings("unchecked") - public void reactiveFullConfig() { - DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfigEndpoint); - Object handler = endpointAccessor.getPropertyValue("handler"); - MessageChannel requestChannel = (MessageChannel) new DirectFieldAccessor( - this.reactiveFullConfigEndpoint).getPropertyValue("inputChannel"); - assertEquals(this.applicationContext.getBean("requests"), requestChannel); - DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); - assertEquals(77, handlerAccessor.getPropertyValue("order")); - assertEquals(Boolean.FALSE, endpointAccessor.getPropertyValue("autoStartup")); - Object replyChannel = handlerAccessor.getPropertyValue("outputChannel"); - assertNotNull(replyChannel); - assertEquals(this.applicationContext.getBean("replies"), replyChannel); - - assertEquals(String.class.getName(), - TestUtils.getPropertyValue(handler, "expectedResponseTypeExpression", Expression.class).getValue()); - Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression"); - assertEquals("http://localhost/test2", uriExpression.getValue()); - assertEquals(HttpMethod.PUT.name(), - TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); - assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); - assertEquals(false, handlerAccessor.getPropertyValue("extractPayload")); - Object sendTimeout = new DirectFieldAccessor( - handlerAccessor.getPropertyValue("messagingTemplate")).getPropertyValue("sendTimeout"); - assertEquals(new Long("1234"), sendTimeout); - Map uriVariableExpressions = - (Map) handlerAccessor.getPropertyValue("uriVariableExpressions"); - assertEquals(1, uriVariableExpressions.size()); - assertEquals("headers.bar", uriVariableExpressions.get("foo").getExpressionString()); - DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("headerMapper")); - String[] mappedRequestHeaders = (String[]) mapperAccessor.getPropertyValue("outboundHeaderNames"); - String[] mappedResponseHeaders = (String[]) mapperAccessor.getPropertyValue("inboundHeaderNames"); - assertEquals(2, mappedRequestHeaders.length); - assertEquals(1, mappedResponseHeaders.length); - assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader1")); - assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2")); - assertEquals("responseHeader", mappedResponseHeaders[0]); - assertEquals(true, handlerAccessor.getPropertyValue("transferCookies")); - } - @Test public void withUrlExpression() { HttpRequestExecutingMessageHandler handler = (HttpRequestExecutingMessageHandler) new DirectFieldAccessor( diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java index 23c646c65a..bfeebbc579 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java @@ -16,9 +16,6 @@ package org.springframework.integration.http.dsl; -import static org.hamcrest.Matchers.instanceOf; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertThat; import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.httpBasic; import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; @@ -30,28 +27,19 @@ import java.util.List; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; -import org.reactivestreams.Publisher; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.ResolvableType; -import org.springframework.http.HttpMethod; -import org.springframework.http.HttpStatus; -import org.springframework.http.MediaType; -import org.springframework.http.client.reactive.ClientHttpConnector; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; -import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; import org.springframework.integration.security.channel.ChannelSecurityInterceptor; import org.springframework.integration.security.channel.SecuredChannel; -import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.PollableChannel; import org.springframework.security.access.AccessDecisionManager; import org.springframework.security.access.vote.AffirmativeBased; import org.springframework.security.access.vote.RoleVoter; @@ -63,20 +51,11 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.test.context.web.WebAppConfiguration; import org.springframework.test.web.client.MockMvcClientHttpRequestFactory; -import org.springframework.test.web.reactive.server.HttpHandlerConnector; -import org.springframework.test.web.reactive.server.WebTestClient; import org.springframework.test.web.servlet.MockMvc; import org.springframework.test.web.servlet.setup.MockMvcBuilders; import org.springframework.util.MultiValueMap; import org.springframework.web.client.RestTemplate; import org.springframework.web.context.WebApplicationContext; -import org.springframework.web.reactive.config.EnableWebFlux; -import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.util.UriComponentsBuilder; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; /** * @author Artem Bilan @@ -95,23 +74,14 @@ public class HttpDslTests { @Autowired private HttpRequestExecutingMessageHandler serviceInternalGatewayHandler; - @Autowired - private ReactiveHttpRequestExecutingMessageHandler serviceInternalReactiveGatewayHandler; - private MockMvc mockMvc; - private WebTestClient webTestClient; - @Before public void setup() { this.mockMvc = MockMvcBuilders.webAppContextSetup(this.wac) .apply(springSecurity()) .build(); - - this.webTestClient = - WebTestClient.bindToApplicationContext(this.wac) - .build(); } @@ -130,71 +100,7 @@ public class HttpDslTests { .string("FOO")); } - @Test - public void testHttpReactiveProxyFlow() throws Exception { - ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { - response.setStatusCode(HttpStatus.OK); - response.getHeaders().setContentType(MediaType.TEXT_PLAIN); - - return response.writeWith(Mono.just(response.bufferFactory().wrap("FOO".getBytes()))) - .then(Mono.defer(response::setComplete)); - }); - - WebClient webClient = WebClient.builder() - .clientConnector(httpConnector) - .build(); - - new DirectFieldAccessor(this.serviceInternalReactiveGatewayHandler) - .setPropertyValue("webClient", webClient); - - this.mockMvc.perform( - get("/service2") - .with(httpBasic("guest", "guest")) - .param("name", "foo")) - .andExpect( - content() - .string("FOO")); - } - - @Autowired - private PollableChannel storeChannel; - - - @Test - @SuppressWarnings("unchecked") - public void testHttpReactivePost() { - this.webTestClient.post().uri("/reactivePost") - .body(Flux.just("foo", "bar", "baz"), String.class) - .exchange() - .expectStatus().isAccepted(); - - Message store = this.storeChannel.receive(10_000); - assertNotNull(store); - assertThat(store.getPayload(), instanceOf(Flux.class)); - - StepVerifier - .create((Publisher) store.getPayload()) - .expectNext("foo", "bar", "baz") - .verifyComplete(); - - } - - @Test - public void testSse() { - Flux responseBody = - this.webTestClient.get().uri("/sse") - .exchange() - .returnResult(String.class) - .getResponseBody(); - - StepVerifier - .create(responseBody) - .expectNext("foo", "bar", "baz") - .verifyComplete(); - } - @Configuration - @EnableWebFlux @EnableWebSecurity @EnableIntegration public static class ContextConfiguration extends WebSecurityConfigurerAdapter { @@ -247,42 +153,6 @@ public class HttpDslTests { .get(); } - @Bean - public IntegrationFlow httpReactiveProxyFlow() { - return IntegrationFlows - .from(Http.inboundGateway("/service2") - .requestMapping(r -> r.params("name"))) - .handle(Http.>outboundReactiveGateway(m -> - UriComponentsBuilder.fromUriString("http://www.springsource.org/spring-integration") - .queryParams(m.getPayload()) - .build() - .toUri()) - .httpMethod(HttpMethod.GET) - .expectedResponseType(String.class), - e -> e.id("serviceInternalReactiveGateway")) - .get(); - } - - @Bean - public IntegrationFlow httpReactiveInboundChannelAdapterFlow() { - return IntegrationFlows - .from(Http.inboundReactiveChannelAdapter("/reactivePost") - .requestMapping(m -> m.methods(HttpMethod.POST)) - .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class)) - .statusCodeFunction(m -> HttpStatus.ACCEPTED)) - .channel(c -> c.queue("storeChannel")) - .get(); - } - - @Bean - public IntegrationFlow sseFlow() { - return IntegrationFlows - .from(Http.inboundReactiveGateway("/sse") - .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) - .handle((p, h) -> Flux.just("foo", "bar", "baz")) - .get(); - } - @Bean public AccessDecisionManager accessDecisionManager() { return new AffirmativeBased(Collections.singletonList(new RoleVoter())); diff --git a/spring-integration-syslog/src/main/resources/org/springframework/integration/syslog/config/spring-integration-syslog-5.0.xsd b/spring-integration-syslog/src/main/resources/org/springframework/integration/syslog/config/spring-integration-syslog-5.0.xsd index 7e58a8c2fa..d661533d0e 100644 --- a/spring-integration-syslog/src/main/resources/org/springframework/integration/syslog/config/spring-integration-syslog-5.0.xsd +++ b/spring-integration-syslog/src/main/resources/org/springframework/integration/syslog/config/spring-integration-syslog-5.0.xsd @@ -1,5 +1,5 @@ -} with 'id' + * {@link WebFluxContextUtils#HANDLER_MAPPING_BEAN_NAME}. + *

+ * In addition, checks if the {@code org.springframework.web.reactive.result.method.RequestMappingInfo} + * class is present on the classpath. + * When Spring Integration HTTP is used only as an HTTP client, there is no reason to use and register + * the HTTP server components. + */ + private void registerReactiveRequestMappingHandlerMappingIfNecessary(BeanDefinitionRegistry registry) { + if (WebFluxContextUtils.WEB_FLUX_PRESENT && + !registry.containsBeanDefinition(WebFluxContextUtils.HANDLER_MAPPING_BEAN_NAME)) { + BeanDefinitionBuilder requestMappingBuilder = + BeanDefinitionBuilder.genericBeanDefinition(WebFluxIntegrationRequestMappingHandlerMapping.class); + requestMappingBuilder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); + requestMappingBuilder.addPropertyValue(IntegrationNamespaceUtils.ORDER, 0); + registry.registerBeanDefinition(WebFluxContextUtils.HANDLER_MAPPING_BEAN_NAME, + requestMappingBuilder.getBeanDefinition()); + + BeanDefinitionReaderUtils.registerWithGeneratedName( + new RootBeanDefinition(IntegrationHandlerResultHandler.class), 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 new file mode 100644 index 0000000000..a97775b646 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java @@ -0,0 +1,35 @@ +/* + * Copyright 2002-2017 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.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; + +/** + * Namespace handler for Spring Integration's webflux namespace. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class WebFluxNamespaceHandler extends AbstractIntegrationNamespaceHandler { + + public void init() { + 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/WebFluxOutboundChannelAdapterParser.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..eb3b69c828 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParser.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-2017 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.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.http.config.HttpOutboundChannelAdapterParser; +import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler; +import org.springframework.util.StringUtils; + +/** + * Parser for the 'outbound-channel-adapter' element of the webflux namespace. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class WebFluxOutboundChannelAdapterParser extends HttpOutboundChannelAdapterParser { + + @Override + protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(WebFluxRequestExecutingMessageHandler.class); + + String webClientRef = element.getAttribute("web-client"); + if (StringUtils.hasText(webClientRef)) { + builder.getBeanDefinition() + .getConstructorArgumentValues() + .addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef)); + } + + return builder; + } + +} 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 new file mode 100644 index 0000000000..0da8b66a1b --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-2017 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.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.http.config.HttpOutboundGatewayParser; +import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler; +import org.springframework.util.StringUtils; + +/** + * Parser for the 'outbound-gateway' element of the webflux namespace. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class WebFluxOutboundGatewayParser extends HttpOutboundGatewayParser { + + @Override + protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(WebFluxRequestExecutingMessageHandler.class); + + String webClientRef = element.getAttribute("web-client"); + if (StringUtils.hasText(webClientRef)) { + builder.getBeanDefinition() + .getConstructorArgumentValues() + .addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef)); + } + + return builder; + } + +} diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/package-info.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/package-info.java new file mode 100644 index 0000000000..81af3146f7 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for configuration - parsers, namespace handlers. + */ +package org.springframework.integration.webflux.config; diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFlux.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFlux.java new file mode 100644 index 0000000000..713a8fd6ba --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFlux.java @@ -0,0 +1,245 @@ +/* + * Copyright 2017 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.dsl; + +import java.net.URI; +import java.util.function.Function; + +import org.springframework.expression.Expression; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint; +import org.springframework.messaging.Message; +import org.springframework.web.reactive.function.client.WebClient; + +/** + * The WebFlux components Factory. + * + * @author Artem Bilan + * @author Shiliang Li + * + * @since 5.0 + */ +public final class WebFlux { + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter based on provided {@link URI}. + * @param uri the {@link URI} to send requests. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundChannelAdapter(URI uri) { + return outboundChannelAdapter(uri, null); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter based on provided {@code uri}. + * @param uri the {@code uri} to send requests. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundChannelAdapter(String uri) { + return outboundChannelAdapter(uri, null); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter based on provided {@code Function} + * to evaluate target {@code uri} against request message. + * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. + * @param

the expected payload type. + * @return the WebFluxMessageHandlerSpec instance + */ + public static

WebFluxMessageHandlerSpec outboundChannelAdapter(Function, ?> uriFunction) { + return outboundChannelAdapter(new FunctionExpression<>(uriFunction)); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter + * based on provided SpEL {@link Expression} to evaluate target {@code uri} + * against request message. + * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundChannelAdapter(Expression uriExpression) { + return outboundChannelAdapter(uriExpression, null); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter + * based on provided {@link URI} and {@link WebClient}. + * @param uri the {@link URI} to send requests. + * @param webClient {@link WebClient} to use. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundChannelAdapter(URI uri, WebClient webClient) { + return new WebFluxMessageHandlerSpec(uri, webClient) + .expectReply(false); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter + * based on provided {@code uri} and {@link WebClient}. + * @param uri the {@code uri} to send requests. + * @param webClient {@link WebClient} to use. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundChannelAdapter(String uri, WebClient webClient) { + return new WebFluxMessageHandlerSpec(uri, webClient) + .expectReply(false); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter + * based on provided {@code Function} to evaluate target {@code uri} against request message + * and {@link WebClient} for HTTP exchanges. + * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. + * @param webClient {@link WebClient} to use. + * @param

the expected payload type. + * @return the WebFluxMessageHandlerSpec instance + */ + public static

WebFluxMessageHandlerSpec outboundChannelAdapter(Function, ?> uriFunction, + WebClient webClient) { + return outboundChannelAdapter(new FunctionExpression<>(uriFunction), webClient); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for one-way adapter + * based on provided SpEL {@link Expression} to evaluate target {@code uri} + * against request message and {@link WebClient} for HTTP exchanges. + * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. + * @param webClient {@link WebClient} to use. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundChannelAdapter(Expression uriExpression, + WebClient webClient) { + return new WebFluxMessageHandlerSpec(uriExpression, webClient) + .expectReply(false); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided {@link URI}. + * @param uri the {@link URI} to send requests. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundGateway(URI uri) { + return outboundGateway(uri, null); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided {@code uri}. + * @param uri the {@code uri} to send requests. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundGateway(String uri) { + return outboundGateway(uri, null); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided {@code Function} to evaluate target {@code uri} against request message. + * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. + * @param

the expected payload type. + * @return the WebFluxMessageHandlerSpec instance + */ + public static

WebFluxMessageHandlerSpec outboundGateway(Function, ?> uriFunction) { + return outboundGateway(new FunctionExpression<>(uriFunction)); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided SpEL {@link Expression} to evaluate target {@code uri} against request message. + * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundGateway(Expression uriExpression) { + return outboundGateway(uriExpression, null); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided {@link URI} and {@link WebClient}. + * @param uri the {@link URI} to send requests. + * @param webClient {@link WebClient} to use. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundGateway(URI uri, WebClient webClient) { + return new WebFluxMessageHandlerSpec(uri, webClient); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided {@code uri} and {@link WebClient}. + * @param uri the {@code uri} to send requests. + * @param webClient {@link WebClient} to use. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundGateway(String uri, WebClient webClient) { + return new WebFluxMessageHandlerSpec(uri, webClient); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided {@code Function} to evaluate target {@code uri} against request message + * and {@link WebClient} for HTTP exchanges. + * @param uriFunction the {@code Function} to evaluate {@code uri} at runtime. + * @param webClient {@link WebClient} to use. + * @param

the expected payload type. + * @return the WebFluxMessageHandlerSpec instance + */ + public static

WebFluxMessageHandlerSpec outboundGateway(Function, ?> uriFunction, + WebClient webClient) { + return outboundGateway(new FunctionExpression<>(uriFunction), webClient); + } + + /** + * Create an {@link WebFluxMessageHandlerSpec} builder for request-reply gateway + * based on provided SpEL {@link Expression} to evaluate target {@code uri} + * against request message and {@link WebClient} for HTTP exchanges. + * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. + * @param webClient {@link WebClient} to use. + * @return the WebFluxMessageHandlerSpec instance + */ + public static WebFluxMessageHandlerSpec outboundGateway(Expression uriExpression, + WebClient webClient) { + return new WebFluxMessageHandlerSpec(uriExpression, webClient); + } + + /** + * Create an {@link WebFluxInboundEndpointSpec} builder for one-way reactive adapter + * based on the provided {@code path} array for mapping. + * @param path the path mapping URIs (e.g. "/myPath.do"). + * @return the WebFluxInboundEndpointSpec instance + */ + public static WebFluxInboundEndpointSpec inboundChannelAdapter(String... path) { + WebFluxInboundEndpoint httpInboundChannelAdapter = new WebFluxInboundEndpoint(false); + return new WebFluxInboundEndpointSpec(httpInboundChannelAdapter, path); + } + + /** + * Create an {@link WebFluxInboundEndpointSpec} builder for request-reply reactive gateway + * based on the provided {@code path} array for mapping. + * @param path the path mapping URIs (e.g. "/myPath.do"). + * @return the WebFluxInboundEndpointSpec instance + */ + public static WebFluxInboundEndpointSpec inboundGateway(String... path) { + return new WebFluxInboundEndpointSpec(new WebFluxInboundEndpoint(), path); + } + + private WebFlux() { + super(); + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java similarity index 63% rename from spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java rename to spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java index b996cf6e17..aac92807c6 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java @@ -14,40 +14,41 @@ * limitations under the License. */ -package org.springframework.integration.http.dsl; +package org.springframework.integration.webflux.dsl; import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.http.codec.ServerCodecConfigurer; -import org.springframework.integration.http.inbound.ReactiveHttpInboundEndpoint; +import org.springframework.integration.http.dsl.HttpInboundEndpointSupportSpec; +import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint; import org.springframework.web.reactive.accept.RequestedContentTypeResolver; /** - * The {@link HttpInboundEndpointSupportSpec} implementation for the {@link ReactiveHttpInboundEndpoint}. + * The {@link HttpInboundEndpointSupportSpec} implementation for the {@link WebFluxInboundEndpoint}. * * @author Artem Bilan * * @since 5.0 */ -public class ReactiveHttpInboundEndpointSpec - extends HttpInboundEndpointSupportSpec { +public class WebFluxInboundEndpointSpec + extends HttpInboundEndpointSupportSpec { - ReactiveHttpInboundEndpointSpec(ReactiveHttpInboundEndpoint gateway, String... path) { + WebFluxInboundEndpointSpec(WebFluxInboundEndpoint gateway, String... path) { super(gateway, path); } - public ReactiveHttpInboundEndpointSpec codecConfigurer(ServerCodecConfigurer codecConfigurer) { + public WebFluxInboundEndpointSpec codecConfigurer(ServerCodecConfigurer codecConfigurer) { this.target.setCodecConfigurer(codecConfigurer); return this; } - public ReactiveHttpInboundEndpointSpec requestedContentTypeResolver( + public WebFluxInboundEndpointSpec requestedContentTypeResolver( RequestedContentTypeResolver requestedContentTypeResolver) { this.target.setRequestedContentTypeResolver(requestedContentTypeResolver); return this; } - public ReactiveHttpInboundEndpointSpec reactiveAdapterRegistry(ReactiveAdapterRegistry adapterRegistry) { + public WebFluxInboundEndpointSpec reactiveAdapterRegistry(ReactiveAdapterRegistry adapterRegistry) { this.target.setReactiveAdapterRegistry(adapterRegistry); return this; } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java similarity index 59% rename from spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java rename to spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java index 898387106c..65aabf3d69 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java @@ -14,41 +14,42 @@ * limitations under the License. */ -package org.springframework.integration.http.dsl; +package org.springframework.integration.webflux.dsl; import java.net.URI; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.expression.ValueExpression; -import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; +import org.springframework.integration.http.dsl.BaseHttpMessageHandlerSpec; +import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler; import org.springframework.web.reactive.function.client.WebClient; /** - * The {@link BaseHttpMessageHandlerSpec} implementation for the {@link ReactiveHttpRequestExecutingMessageHandler}. + * The {@link BaseHttpMessageHandlerSpec} implementation for the {@link WebFluxRequestExecutingMessageHandler}. * * @author Shiliang Li * @author Artem Bilan * * @since 5.0 * - * @see ReactiveHttpRequestExecutingMessageHandler + * @see WebFluxRequestExecutingMessageHandler */ -public class ReactiveHttpMessageHandlerSpec - extends BaseHttpMessageHandlerSpec { +public class WebFluxMessageHandlerSpec + extends BaseHttpMessageHandlerSpec { private final WebClient webClient; - ReactiveHttpMessageHandlerSpec(URI uri, WebClient webClient) { + WebFluxMessageHandlerSpec(URI uri, WebClient webClient) { this(new ValueExpression<>(uri), webClient); } - ReactiveHttpMessageHandlerSpec(String uri, WebClient webClient) { + WebFluxMessageHandlerSpec(String uri, WebClient webClient) { this(new LiteralExpression(uri), webClient); } - ReactiveHttpMessageHandlerSpec(Expression uriExpression, WebClient webClient) { - super(new ReactiveHttpRequestExecutingMessageHandler(uriExpression, webClient)); + WebFluxMessageHandlerSpec(Expression uriExpression, WebClient webClient) { + super(new WebFluxRequestExecutingMessageHandler(uriExpression, webClient)); this.webClient = webClient; } @@ -57,4 +58,9 @@ public class ReactiveHttpMessageHandlerSpec return this.webClient != null; } + @Override + protected WebFluxMessageHandlerSpec expectReply(boolean expectReply) { + return super.expectReply(expectReply); + } + } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/package-info.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/package-info.java new file mode 100644 index 0000000000..965db424f4 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides WebFlux Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.webflux.dsl; diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java similarity index 85% rename from spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java rename to spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java index 5488cfb696..98ccef31f2 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.http.inbound; +package org.springframework.integration.webflux.inbound; import org.springframework.core.Ordered; import org.springframework.web.method.HandlerMethod; @@ -26,14 +26,14 @@ import reactor.core.publisher.Mono; /** * A {@link HandlerResultHandler} implementation to handle the result of the - * {@link ReactiveHttpInboundEndpoint} execution. Actually just return the + * {@link WebFluxInboundEndpoint} execution. Actually just return the * {@code result.getReturnValue()} which essentially is expected {@code Mono}. * * @author Artem Bilan * * @since 5.0 * - * @see ReactiveHttpInboundEndpoint + * @see WebFluxInboundEndpoint */ public class IntegrationHandlerResultHandler implements HandlerResultHandler, Ordered { @@ -41,7 +41,7 @@ public class IntegrationHandlerResultHandler implements HandlerResultHandler, Or public boolean supports(HandlerResult result) { Object handler = result.getHandler(); return handler instanceof HandlerMethod - && ReactiveHttpInboundEndpoint.class.isAssignableFrom(((HandlerMethod) handler).getBeanType()); + && WebFluxInboundEndpoint.class.isAssignableFrom(((HandlerMethod) handler).getBeanType()); } @Override diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java similarity index 97% rename from spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java rename to spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index df25d49293..0bb4f2a526 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.http.inbound; +package org.springframework.integration.webflux.inbound; import java.util.ArrayList; import java.util.Collections; @@ -43,6 +43,7 @@ import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.http.inbound.BaseHttpInboundEndpoint; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -70,7 +71,7 @@ import reactor.core.publisher.Mono; * @see org.springframework.web.reactive.result.HandlerResultHandlerSupport * @see org.springframework.web.reactive.config.EnableWebFlux */ -public class ReactiveHttpInboundEndpoint extends BaseHttpInboundEndpoint implements WebHandler { +public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements WebHandler { private static final MediaType MEDIA_TYPE_APPLICATION_ALL = new MediaType("application"); @@ -80,11 +81,11 @@ public class ReactiveHttpInboundEndpoint extends BaseHttpInboundEndpoint impleme private ReactiveAdapterRegistry adapterRegistry = new ReactiveAdapterRegistry(); - public ReactiveHttpInboundEndpoint() { + public WebFluxInboundEndpoint() { this(true); } - public ReactiveHttpInboundEndpoint(boolean expectReply) { + public WebFluxInboundEndpoint(boolean expectReply) { super(expectReply); } @@ -119,7 +120,7 @@ public class ReactiveHttpInboundEndpoint extends BaseHttpInboundEndpoint impleme @Override public String getComponentType() { - return super.getComponentType().replaceFirst("(http:)", "$1reactive-"); + return super.getComponentType().replaceFirst("(http:)", "$1webflux-"); } @Override diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java similarity index 76% rename from spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java rename to spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java index 5edc84883d..7e35f7d849 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java @@ -14,14 +14,18 @@ * limitations under the License. */ -package org.springframework.integration.http.inbound; +package org.springframework.integration.webflux.inbound; import java.lang.reflect.Method; import java.util.concurrent.atomic.AtomicBoolean; import org.springframework.context.ApplicationListener; import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.integration.http.inbound.BaseHttpInboundEndpoint; +import org.springframework.integration.http.inbound.CrossOrigin; +import org.springframework.integration.http.inbound.RequestMapping; import org.springframework.integration.http.support.HttpContextUtils; +import org.springframework.integration.webflux.support.WebFluxContextUtils; import org.springframework.util.CollectionUtils; import org.springframework.util.ReflectionUtils; import org.springframework.web.bind.annotation.RequestMethod; @@ -33,18 +37,19 @@ import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebHandler; /** - * The {@link org.springframework.web.servlet.HandlerMapping} implementation that - * detects and registers {@link org.springframework.web.servlet.mvc.method.RequestMappingInfo}s for - * {@link HttpRequestHandlingEndpointSupport} from a Spring Integration HTTP configuration + * The {@link org.springframework.web.reactive.HandlerMapping} implementation that + * detects and registers {@link org.springframework.web.reactive.result.method.RequestMappingInfo}s for + * {@link org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport} + * from a Spring Integration HTTP configuration * of {@code } and {@code } elements. *

* This class is automatically configured as a bean in the application context during the - * parsing phase of the {@code } + * parsing phase of the {@code } * elements, if there is none registered, yet. However it can be configured as a regular * bean with appropriate configuration for * {@link org.springframework.web.reactive.result.method.annotation.RequestMappingHandlerMapping}. * It is recommended to have only one similar bean in the application context using the 'id' - * {@link org.springframework.integration.http.support.HttpContextUtils#REACTIVE_HANDLER_MAPPING_BEAN_NAME}. + * {@link WebFluxContextUtils#HANDLER_MAPPING_BEAN_NAME}. *

* In most cases, Spring MVC offers to configure Request Mapping via * {@code org.springframework.stereotype.Controller} and @@ -54,10 +59,10 @@ import org.springframework.web.server.WebHandler; * {@code org.springframework.stereotype.Controller} user-class may have their own * {@link org.springframework.web.bind.annotation.RequestMapping}. * On the other side, all Spring Integration HTTP Inbound Endpoints are configured on - * the basis of the same {@link HttpRequestHandlingEndpointSupport} class and there is no - * single {@link org.springframework.web.reactive.result.method.RequestMappingInfo} configuration without - * {@link org.springframework.web.method.HandlerMethod} in Spring MVC. - * Accordingly {@link ReactiveIntegrationRequestMappingHandlerMapping} is a + * the basis of the same {@link org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport} + * class and there is no single {@link org.springframework.web.reactive.result.method.RequestMappingInfo} + * configuration without {@link org.springframework.web.method.HandlerMethod} in Spring MVC. + * Accordingly {@link WebFluxIntegrationRequestMappingHandlerMapping} is a * {@link org.springframework.web.reactive.HandlerMapping} * compromise implementation between method-level annotations and component-level * (e.g. Spring Integration XML) configurations. @@ -69,7 +74,7 @@ import org.springframework.web.server.WebHandler; * @see RequestMapping * @see RequestMappingHandlerMapping */ -public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMappingHandlerMapping +public class WebFluxIntegrationRequestMappingHandlerMapping extends RequestMappingHandlerMapping implements ApplicationListener { private static final Method HANDLER_METHOD = ReflectionUtils.findMethod(WebHandler.class, @@ -79,7 +84,7 @@ public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMapp @Override protected boolean isHandler(Class beanType) { - return ReactiveHttpInboundEndpoint.class.isAssignableFrom(beanType); + return WebFluxInboundEndpoint.class.isAssignableFrom(beanType); } @Override @@ -87,7 +92,7 @@ public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMapp if (handler instanceof String) { handler = getApplicationContext().getBean((String) handler); } - RequestMappingInfo mapping = getMappingForEndpoint((ReactiveHttpInboundEndpoint) handler); + RequestMappingInfo mapping = getMappingForEndpoint((WebFluxInboundEndpoint) handler); if (mapping != null) { registerMapping(mapping, handler, HANDLER_METHOD); } @@ -98,7 +103,7 @@ public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMapp * a Spring Integration Reactive HTTP Inbound Endpoint {@link RequestMapping}. * @see RequestMappingHandlerMapping#getMappingForMethod */ - private RequestMappingInfo getMappingForEndpoint(ReactiveHttpInboundEndpoint endpoint) { + private RequestMappingInfo getMappingForEndpoint(WebFluxInboundEndpoint endpoint) { org.springframework.web.bind.annotation.RequestMapping requestMappingAnnotation = HttpContextUtils.convertRequestMappingToAnnotation(endpoint.getRequestMapping()); if (requestMappingAnnotation != null) { @@ -149,16 +154,11 @@ public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMapp return null; } - @Override - public void afterPropertiesSet() { - // No-op in favor of onApplicationEvent - } - /** - * {@link HttpRequestHandlingEndpointSupport}s may depend on auto-created - * {@code requestChannel}s, so MVC Handlers detection should be postponed + * {@link org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport}s + * may depend on auto-created {@code requestChannel}s, so MVC Handlers detection should be postponed * as late as possible. - * @see org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping#afterPropertiesSet() + * @see RequestMappingHandlerMapping#afterPropertiesSet() */ @Override public void onApplicationEvent(ContextRefreshedEvent event) { @@ -167,4 +167,9 @@ public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMapp } } + @Override + public void afterPropertiesSet() { + // No-op in favor of onApplicationEvent + } + } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/package-info.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/package-info.java new file mode 100644 index 0000000000..5dd7f19ba7 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes supporting inbound endpoints. + */ +package org.springframework.integration.webflux.inbound; diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java similarity index 87% rename from spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java rename to spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java index 5f372a6f1f..3677f0c5c4 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.http.outbound; +package org.springframework.integration.webflux.outbound; import java.net.URI; import java.nio.charset.StandardCharsets; @@ -32,6 +32,7 @@ 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; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -54,9 +55,9 @@ import reactor.core.publisher.Mono; * * @since 5.0 * - * @see HttpRequestExecutingMessageHandler + * @see org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler */ -public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequestExecutingMessageHandler { +public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestExecutingMessageHandler { private final WebClient webClient; @@ -64,7 +65,7 @@ public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequ * Create a handler that will send requests to the provided URI. * @param uri The URI. */ - public ReactiveHttpRequestExecutingMessageHandler(URI uri) { + public WebFluxRequestExecutingMessageHandler(URI uri) { this(new ValueExpression<>(uri)); } @@ -72,7 +73,7 @@ public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequ * Create a handler that will send requests to the provided URI. * @param uri The URI. */ - public ReactiveHttpRequestExecutingMessageHandler(String uri) { + public WebFluxRequestExecutingMessageHandler(String uri) { this(uri, null); } @@ -80,7 +81,7 @@ public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequ * Create a handler that will send requests to the provided URI Expression. * @param uriExpression The URI expression. */ - public ReactiveHttpRequestExecutingMessageHandler(Expression uriExpression) { + public WebFluxRequestExecutingMessageHandler(Expression uriExpression) { this(uriExpression, null); } @@ -89,7 +90,7 @@ public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequ * @param uri The URI. * @param webClient The WebClient to use. */ - public ReactiveHttpRequestExecutingMessageHandler(String uri, WebClient webClient) { + public WebFluxRequestExecutingMessageHandler(String uri, WebClient webClient) { this(new LiteralExpression(uri), webClient); /* * We'd prefer to do this assertion first, but the compiler doesn't allow it. However, @@ -105,7 +106,7 @@ public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequ * {@link BeanFactory}. * @param webClient The WebClient to use. */ - public ReactiveHttpRequestExecutingMessageHandler(Expression uriExpression, WebClient webClient) { + public WebFluxRequestExecutingMessageHandler(Expression uriExpression, WebClient webClient) { super(uriExpression); this.webClient = (webClient == null ? WebClient.create() : webClient); this.setAsync(true); @@ -113,7 +114,7 @@ public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequ @Override public String getComponentType() { - return (isExpectReply() ? "http:outbound-reactive-gateway" : "http:outbound-reactive-channel-adapter"); + return (isExpectReply() ? "webflux:outbound-gateway" : "webflux:outbound-channel-adapter"); } @Override diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/package-info.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/package-info.java new file mode 100644 index 0000000000..8aaaa48992 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes supporting outbound endpoints. + */ +package org.springframework.integration.webflux.outbound; diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/WebFluxContextUtils.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/WebFluxContextUtils.java new file mode 100644 index 0000000000..8e92602fd8 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/WebFluxContextUtils.java @@ -0,0 +1,50 @@ +/* + * Copyright 2017 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.util.ClassUtils; + +/** + * Utility class for accessing WebFlux integration components + * from the {@link org.springframework.beans.factory.BeanFactory}. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public final class WebFluxContextUtils { + + private WebFluxContextUtils() { + super(); + } + + /** + * The {@code boolean} flag to indicate if the + * {@code org.springframework.web.reactive.result.method.RequestMappingInfo} + * is present in the CLASSPATH to allow to register the Integration server reactive components. + */ + public static final boolean WEB_FLUX_PRESENT = + ClassUtils.isPresent("org.springframework.web.reactive.result.method.RequestMappingInfo", + WebFluxContextUtils.class.getClassLoader()); + + /** + * The name for the infrastructure + * {@link org.springframework.integration.webflux.inbound.WebFluxIntegrationRequestMappingHandlerMapping} bean. + */ + public static final String HANDLER_MAPPING_BEAN_NAME = "webFluxIntegrationRequestMappingHandlerMapping"; + +} diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/package-info.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/package-info.java new file mode 100644 index 0000000000..e9427f0978 --- /dev/null +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes to support WebFlux endpoints. + */ +package org.springframework.integration.webflux.support; diff --git a/spring-integration-webflux/src/main/resources/META-INF/spring.factories b/spring-integration-webflux/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000000..75988e5b4d --- /dev/null +++ b/spring-integration-webflux/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.integration.config.IntegrationConfigurationInitializer=\ +org.springframework.integration.webflux.config.WebFluxIntegrationConfigurationInitializer diff --git a/spring-integration-webflux/src/main/resources/META-INF/spring.handlers b/spring-integration-webflux/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..94ef26da00 --- /dev/null +++ b/spring-integration-webflux/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/webflux=org.springframework.integration.webflux.config.WebFluxNamespaceHandler \ No newline at end of file diff --git a/spring-integration-webflux/src/main/resources/META-INF/spring.schemas b/spring-integration-webflux/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..490f885418 --- /dev/null +++ b/spring-integration-webflux/src/main/resources/META-INF/spring.schemas @@ -0,0 +1,2 @@ +http\://www.springframework.org/schema/integration/webflux/spring-integration-webflux-5.0.xsd=org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd +http\://www.springframework.org/schema/integration/webflux/spring-integration-webflux.xsd=org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd diff --git a/spring-integration-webflux/src/main/resources/META-INF/spring.tooling b/spring-integration-webflux/src/main/resources/META-INF/spring.tooling new file mode 100644 index 0000000000..227b56225f --- /dev/null +++ b/spring-integration-webflux/src/main/resources/META-INF/spring.tooling @@ -0,0 +1,4 @@ +# Tooling related information for the integration webflux namespace +http\://www.springframework.org/schema/integration/webflux@name=integration webflux Namespace +http\://www.springframework.org/schema/integration/webflux@prefix=int-webflux +http\://www.springframework.org/schema/integration/webflux@icon=org/springframework/integration/webflux/config/spring-integration-webflux.gif 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 new file mode 100644 index 0000000000..4397635b54 --- /dev/null +++ b/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd @@ -0,0 +1,181 @@ + + + + + + + + + + + + + + + + Configures a Consumer Endpoint for the + 'org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler' + with 'expectReply = false' that sends HTTP requests based on incoming messages reactive manner. + + + + + + + + Specify an expression for URI variable placeholder within 'url'. + This element is mutually exclusive with 'uri-variables-expression' attribute. + + + + + + + + + + + + + + + + + A reference to an org.springframework.web.reactive.function.client.WebClient bean + which is used to send to send the HTTP Requests reactive manner. + + + + + + + Specify whether the outbound message's payload should be extracted + when preparing the request body. Otherwise the Message instance itself + will be serialized. + The default value is 'true'. + + + + + + + + + + Configures a Consumer Endpoint for the + 'org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler' + that sends HTTP requests based on incoming messages and expects HTTP responses. + + + + + + + + + + Specify an expression for URI variable placeholder within 'url'. + + + + + + + + + + + + + + + + The receiving Message Channel of this endpoint. + + + + + + + + + + + + Specifies whether the outbound message's payload should be extracted + when preparing the request body. Otherwise the Message instance itself + will be serialized. + The default value is 'true'. + + + + + + + + + + + + + + + + + + + + + + + A reference to an org.springframework.web.reactive.function.client.WebClient bean + which is used to send to send the HTTP Requests reactive manner. + + + + + + + + + diff --git a/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux.gif b/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux.gif new file mode 100644 index 0000000000000000000000000000000000000000..750667e608fcaf4e40253ddf3505d59c2bdfcb44 GIT binary patch literal 578 zcmV-I0=@l5Nk%w1VGsZi0OkMy00030|NnVQPjYW|Wnyk~b9G-}dtGRXTXMBraiCav zyjgj5Kt;tik*L|a|uF=~>vd?g}#YDHzV2F`@y~K0D#e2uhjKRf}xwW*~)QiKor@*_l z?$(5^oZ#x?((BIj{Oq&ReW=84gOF&XxOS$jmWhgdo19Cjt7@>Xn5U+aiHLc8dR?!q zk?7yC;@!l|y^Gkyj?ur2>DHj<;LN_ER@uak;Lf7!-oNg`V(ihB?Z=P$Rm6Q$pJqk2bx_8D zSh;;%YCb2mbW^f)RI6=JvTj(ZY+9#fR+(H;mQ+&z|Nj6000000A^8LW004aeEC2ui z01yBW000M{fPI34goTEOh<#2}R!xb5XJ158Z(WH`L}YndVsTrDQ(j9~YH4V2ZHHKR zV@zOsb8=;eMto~-N_2X5Y=~KAb$2c)FDfV;hg@MAAW=nCEh8B=h7Bb#P&Gd*8xuFx zA}~NRGC3p*27M6_4ebpbCnhvFJqiy6eMv}1LPJ3;Av|CZAR)m3AL`h-yYPWR1q2KT Q?096s!2upR1O);BJ6#qKw*UYD literal 0 HcmV?d00001 diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..8b7635616a --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests-context.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..9af4f6505b --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundChannelAdapterParserTests.java @@ -0,0 +1,88 @@ +/* + * Copyright 2017 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.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; + +import java.nio.charset.Charset; + +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.context.ApplicationContext; +import org.springframework.expression.Expression; +import org.springframework.http.HttpMethod; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.web.reactive.function.client.WebClient; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class HttpOutboundChannelAdapterParserTests { + + @Autowired + @Qualifier("reactiveMinimalConfig") + private AbstractEndpoint reactiveMinimalConfig; + + @Autowired + @Qualifier("reactiveWebClientConfig") + private AbstractEndpoint reactiveWebClientConfig; + + @Autowired + private WebClient webClient; + + @Autowired + private ApplicationContext applicationContext; + + @Test + public void reactiveMinimalConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveMinimalConfig); + WebClient webClient = + TestUtils.getPropertyValue(this.reactiveMinimalConfig, "handler.webClient", WebClient.class); + assertNotSame(this.webClient, webClient); + Object handler = endpointAccessor.getPropertyValue("handler"); + DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); + assertEquals(false, handlerAccessor.getPropertyValue("expectReply")); + assertEquals(this.applicationContext.getBean("requests"), endpointAccessor.getPropertyValue("inputChannel")); + assertNull(handlerAccessor.getPropertyValue("outputChannel")); + Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression"); + assertEquals("http://localhost/test1", uriExpression.getValue()); + assertEquals(HttpMethod.POST.name(), + TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); + assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); + assertEquals(true, handlerAccessor.getPropertyValue("extractPayload")); + } + + @Test + public void reactiveWebClientConfig() { + assertSame(this.webClient, TestUtils.getPropertyValue(this.reactiveWebClientConfig, "handler.webClient")); + } + +} diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests-context.xml new file mode 100644 index 0000000000..3a4b996022 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests-context.xml @@ -0,0 +1,42 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests.java new file mode 100644 index 0000000000..2c229aa733 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/HttpOutboundGatewayParserTests.java @@ -0,0 +1,128 @@ +/* + * Copyright 2017 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.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.nio.charset.Charset; +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.context.ApplicationContext; +import org.springframework.expression.Expression; +import org.springframework.http.HttpMethod; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.test.util.TestUtils; +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.client.WebClient; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class HttpOutboundGatewayParserTests { + + @Autowired + @Qualifier("reactiveMinimalConfig") + private AbstractEndpoint reactiveMinimalConfigEndpoint; + + @Autowired + @Qualifier("reactiveFullConfig") + private AbstractEndpoint reactiveFullConfigEndpoint; + + @Autowired + private WebClient webClient; + + @Autowired + private ApplicationContext applicationContext; + + @Test + public void reactiveMinimalConfig() { + Object handler = new DirectFieldAccessor(this.reactiveMinimalConfigEndpoint).getPropertyValue("handler"); + Object requestChannel = new DirectFieldAccessor(this.reactiveMinimalConfigEndpoint) + .getPropertyValue("inputChannel"); + assertEquals(this.applicationContext.getBean("requests"), requestChannel); + DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); + Object replyChannel = handlerAccessor.getPropertyValue("outputChannel"); + assertNull(replyChannel); + assertSame(this.webClient, handlerAccessor.getPropertyValue("webClient")); + Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression"); + assertEquals("http://localhost/test1", uriExpression.getValue()); + assertEquals(HttpMethod.POST.name(), + TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); + assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); + assertEquals(true, handlerAccessor.getPropertyValue("extractPayload")); + assertEquals(false, handlerAccessor.getPropertyValue("transferCookies")); + } + + @Test + @SuppressWarnings("unchecked") + public void reactiveFullConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfigEndpoint); + Object handler = endpointAccessor.getPropertyValue("handler"); + MessageChannel requestChannel = (MessageChannel) new DirectFieldAccessor( + this.reactiveFullConfigEndpoint).getPropertyValue("inputChannel"); + assertEquals(this.applicationContext.getBean("requests"), requestChannel); + DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); + assertEquals(77, handlerAccessor.getPropertyValue("order")); + assertEquals(Boolean.FALSE, endpointAccessor.getPropertyValue("autoStartup")); + Object replyChannel = handlerAccessor.getPropertyValue("outputChannel"); + assertNotNull(replyChannel); + assertEquals(this.applicationContext.getBean("replies"), replyChannel); + + assertEquals(String.class.getName(), + TestUtils.getPropertyValue(handler, "expectedResponseTypeExpression", Expression.class).getValue()); + Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression"); + assertEquals("http://localhost/test2", uriExpression.getValue()); + assertEquals(HttpMethod.PUT.name(), + TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); + assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); + assertEquals(false, handlerAccessor.getPropertyValue("extractPayload")); + Object sendTimeout = new DirectFieldAccessor( + handlerAccessor.getPropertyValue("messagingTemplate")).getPropertyValue("sendTimeout"); + assertEquals(new Long("1234"), sendTimeout); + Map uriVariableExpressions = + (Map) handlerAccessor.getPropertyValue("uriVariableExpressions"); + assertEquals(1, uriVariableExpressions.size()); + assertEquals("headers.bar", uriVariableExpressions.get("foo").getExpressionString()); + DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("headerMapper")); + String[] mappedRequestHeaders = (String[]) mapperAccessor.getPropertyValue("outboundHeaderNames"); + String[] mappedResponseHeaders = (String[]) mapperAccessor.getPropertyValue("inboundHeaderNames"); + assertEquals(2, mappedRequestHeaders.length); + assertEquals(1, mappedResponseHeaders.length); + assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader1")); + assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2")); + assertEquals("responseHeader", mappedResponseHeaders[0]); + assertEquals(true, handlerAccessor.getPropertyValue("transferCookies")); + } + +} 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 new file mode 100644 index 0000000000..f533550ec2 --- /dev/null +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java @@ -0,0 +1,237 @@ +/* + * Copyright 2016-2017 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.dsl; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.httpBasic; +import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; + +import java.util.Collections; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.ResolvableType; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.client.reactive.ClientHttpConnector; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.http.dsl.Http; +import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.security.access.AccessDecisionManager; +import org.springframework.security.access.vote.AffirmativeBased; +import org.springframework.security.access.vote.RoleVoter; +import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder; +import org.springframework.security.config.annotation.web.builders.HttpSecurity; +import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity; +import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.test.web.reactive.server.HttpHandlerConnector; +import org.springframework.test.web.reactive.server.WebTestClient; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import org.springframework.util.MultiValueMap; +import org.springframework.web.context.WebApplicationContext; +import org.springframework.web.reactive.config.EnableWebFlux; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.util.UriComponentsBuilder; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author Artem Bilan + * @author Shiliang Li + * + * @since 5.0 + */ +@RunWith(SpringRunner.class) +@WebAppConfiguration +@DirtiesContext +public class WebFluxDslTests { + + @Autowired + private WebApplicationContext wac; + + @Autowired + private WebFluxRequestExecutingMessageHandler serviceInternalReactiveGatewayHandler; + + private MockMvc mockMvc; + + private WebTestClient webTestClient; + + @Before + public void setup() { + this.mockMvc = + MockMvcBuilders.webAppContextSetup(this.wac) + .apply(springSecurity()) + .build(); + + this.webTestClient = + WebTestClient.bindToApplicationContext(this.wac) + .build(); + } + + @Test + public void testHttpReactiveProxyFlow() throws Exception { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + response.setStatusCode(HttpStatus.OK); + response.getHeaders().setContentType(MediaType.TEXT_PLAIN); + + return response.writeWith(Mono.just(response.bufferFactory().wrap("FOO".getBytes()))) + .then(Mono.defer(response::setComplete)); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + new DirectFieldAccessor(this.serviceInternalReactiveGatewayHandler) + .setPropertyValue("webClient", webClient); + + this.mockMvc.perform( + get("/service2") + .with(httpBasic("guest", "guest")) + .param("name", "foo")) + .andExpect( + content() + .string("FOO")); + } + + @Autowired + private PollableChannel storeChannel; + + + @Test + @SuppressWarnings("unchecked") + public void testHttpReactivePost() { + this.webTestClient.post().uri("/reactivePost") + .body(Flux.just("foo", "bar", "baz"), String.class) + .exchange() + .expectStatus().isAccepted(); + + Message store = this.storeChannel.receive(10_000); + assertNotNull(store); + assertThat(store.getPayload(), instanceOf(Flux.class)); + + StepVerifier + .create((Publisher) store.getPayload()) + .expectNext("foo", "bar", "baz") + .verifyComplete(); + + } + + @Test + public void testSse() { + Flux responseBody = + this.webTestClient.get().uri("/sse") + .exchange() + .returnResult(String.class) + .getResponseBody(); + + StepVerifier + .create(responseBody) + .expectNext("foo", "bar", "baz") + .verifyComplete(); + } + + @Configuration + @EnableWebFlux + @EnableWebSecurity + @EnableIntegration + public static class ContextConfiguration extends WebSecurityConfigurerAdapter { + + @Override + protected void configure(AuthenticationManagerBuilder auth) throws Exception { + auth.inMemoryAuthentication() + .withUser("guest") + .password("guest") + .roles("ADMIN"); + } + + @Override + protected void configure(HttpSecurity http) throws Exception { + http.authorizeRequests() + .anyRequest().hasRole("ADMIN") + .and() + .httpBasic() + .and() + .csrf().disable() + .anonymous().disable(); + } + + @Bean + public IntegrationFlow httpReactiveProxyFlow() { + return IntegrationFlows + .from(Http.inboundGateway("/service2") + .requestMapping(r -> r.params("name"))) + .handle(WebFlux.>outboundGateway(m -> + UriComponentsBuilder.fromUriString("http://www.springsource.org/spring-integration") + .queryParams(m.getPayload()) + .build() + .toUri()) + .httpMethod(HttpMethod.GET) + .expectedResponseType(String.class)) + .get(); + } + + @Bean + public IntegrationFlow httpReactiveInboundChannelAdapterFlow() { + return IntegrationFlows + .from(WebFlux.inboundChannelAdapter("/reactivePost") + .requestMapping(m -> m.methods(HttpMethod.POST)) + .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class)) + .statusCodeFunction(m -> HttpStatus.ACCEPTED)) + .channel(c -> c.queue("storeChannel")) + .get(); + } + + @Bean + public IntegrationFlow sseFlow() { + return IntegrationFlows + .from(WebFlux.inboundGateway("/sse") + .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) + .handle((p, h) -> Flux.just("foo", "bar", "baz")) + .get(); + } + + @Bean + public AccessDecisionManager accessDecisionManager() { + return new AffirmativeBased(Collections.singletonList(new RoleVoter())); + } + + } + +} diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java similarity index 90% rename from spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java rename to spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java index 2c631fb796..c930535d84 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.http.inbound; +package org.springframework.integration.webflux.inbound; import java.util.Objects; @@ -30,6 +30,7 @@ import org.springframework.http.MediaType; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.http.inbound.RequestMapping; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; @@ -47,13 +48,13 @@ import reactor.core.publisher.Flux; */ @RunWith(SpringRunner.class) @DirtiesContext -public class ReactiveHttpInboundEndpointTests { +public class WebFluxInboundEndpointTests { @Autowired private WebTestClient webTestClient; @Autowired - private ReactiveHttpInboundEndpoint simpleInboundEndpoint; + private WebFluxInboundEndpoint simpleInboundEndpoint; @Test public void testSimpleGet() { @@ -93,8 +94,8 @@ public class ReactiveHttpInboundEndpointTests { } @Bean - public ReactiveHttpInboundEndpoint simpleInboundEndpoint() { - ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); + public WebFluxInboundEndpoint simpleInboundEndpoint() { + WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint(); RequestMapping requestMapping = new RequestMapping(); requestMapping.setPathPatterns("/test"); endpoint.setRequestMapping(requestMapping); @@ -108,8 +109,8 @@ public class ReactiveHttpInboundEndpointTests { } @Bean - public ReactiveHttpInboundEndpoint jsonInboundEndpoint() { - ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); + public WebFluxInboundEndpoint jsonInboundEndpoint() { + WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint(); RequestMapping requestMapping = new RequestMapping(); requestMapping.setPathPatterns("/persons"); endpoint.setRequestMapping(requestMapping); diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java similarity index 87% rename from spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java rename to spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java index 029a599e21..d68a07b5d4 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.http.outbound; +package org.springframework.integration.webflux.outbound; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; @@ -48,14 +48,13 @@ import reactor.test.StepVerifier; * * @since 5.0 */ -public class ReactiveHttpRequestExecutingMessageHandlerTests { +public class WebFluxRequestExecutingMessageHandlerTests { @Test public void testReactiveReturn() throws Throwable { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.OK); - return Mono.empty() - .then(Mono.defer(response::setComplete)); + return Mono.defer(response::setComplete); }); WebClient webClient = WebClient.builder() @@ -63,8 +62,8 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { .build(); String destinationUri = "http://www.springsource.org/spring-integration"; - ReactiveHttpRequestExecutingMessageHandler reactiveHandler = - new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); + WebFluxRequestExecutingMessageHandler reactiveHandler = + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); FluxMessageChannel ackChannel = new FluxMessageChannel(); reactiveHandler.setOutputChannel(ackChannel); @@ -84,8 +83,7 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { public void testReactiveErrorOneWay() throws Throwable { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.UNAUTHORIZED); - return Mono.empty() - .then(Mono.defer(response::setComplete)); + return Mono.defer(response::setComplete); }); WebClient webClient = WebClient.builder() @@ -93,8 +91,8 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { .build(); String destinationUri = "http://www.springsource.org/spring-integration"; - ReactiveHttpRequestExecutingMessageHandler reactiveHandler = - new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); + WebFluxRequestExecutingMessageHandler reactiveHandler = + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); reactiveHandler.setExpectReply(false); QueueChannel errorChannel = new QueueChannel(); @@ -121,8 +119,8 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { .build(); String destinationUri = "http://www.springsource.org/spring-integration"; - ReactiveHttpRequestExecutingMessageHandler reactiveHandler = - new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); + WebFluxRequestExecutingMessageHandler reactiveHandler = + new WebFluxRequestExecutingMessageHandler(destinationUri, webClient); reactiveHandler.setExpectReply(false); QueueChannel errorChannel = new QueueChannel(); @@ -138,5 +136,4 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { assertThat(throwable.getMessage(), containsString("Intentional connection error")); } - } diff --git a/spring-integration-webflux/src/test/resources/log4j.properties b/spring-integration-webflux/src/test/resources/log4j.properties new file mode 100644 index 0000000000..c65576b50f --- /dev/null +++ b/spring-integration-webflux/src/test/resources/log4j.properties @@ -0,0 +1,9 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n + +#log4j.category.org.springframework=DEBUG +log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration.webflux=WARN diff --git a/src/reference/asciidoc/http.adoc b/src/reference/asciidoc/http.adoc index a04caa5a3f..713246e336 100644 --- a/src/reference/asciidoc/http.adoc +++ b/src/reference/asciidoc/http.adoc @@ -5,7 +5,8 @@ === Introduction The HTTP support allows for the execution of HTTP requests and the processing of inbound HTTP requests. -the HTTP support consists of the following gateway implementations: `HttpInboundEndpoint`, `HttpRequestExecutingMessageHandler` and `ReactiveHttpRequestExecutingMessageHandler` +The HTTP support consists of the following gateway implementations: `HttpInboundEndpoint`, `HttpRequestExecutingMessageHandler`. +Also see <>. [[http-inbound]] === Http Inbound Components @@ -122,59 +123,6 @@ This also shows how to customize the HTTP methods accepted by the gateway, which The reply message will be available in the Model map. The key that is used for that map entry by default is 'reply', but this can be overridden by setting the 'replyKey' property on the endpoint's configuration. -=== WebFlux Server Side support - -Starting with _version 5.0_, the `ReactiveHttpInboundEndpoint`, http://docs.spring.io/spring/docs/5.0.0.RC3/spring-framework-reference/web.html#web-reactive[WebFlux] `WebHandler`, implementation is provided. -This component is similar to the MVC-based `HttpRequestHandlingEndpointSupport` with which it shares some common options via the newly extracted `BaseHttpInboundEndpoint`. -Instead of MVC, it is used in the Spring WebFlux Reactive environment. -A simple sample for explanation: - -[source,java] ----- -@Configuration -@EnableWebFlux -@EnableIntegration -public class ReactiveHttpConfiguration { - - @Bean - public ReactiveHttpInboundEndpoint simpleInboundEndpoint() { - ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); - RequestMapping requestMapping = new RequestMapping(); - requestMapping.setPathPatterns("/test"); - endpoint.setRequestMapping(requestMapping); - endpoint.setRequestChannelName("serviceChannel"); - return endpoint; - } - - @ServiceActivator(inputChannel = "serviceChannel") - String service() { - return "It works!"; - } - -} ----- - -As can be seen, the configuration is similar to the `HttpRequestHandlingEndpointSupport` mentioned above, except that we use `@EnableWebFlux` to add the WebFlux infrastructure to our integration application. -Also, the `ReactiveHttpInboundEndpoint` performs `sendAndReceive` operation to the downstream flow using back-pressure, on demand based capabilities, provided by the reactive HTTP server implementation. - -NOTE: The reply part is non-blocking as well and based on the internal `FutureReplyChannel` which is flat-mapped to a reply `Mono` for on demand resolution. - -The `ReactiveHttpInboundEndpoint` can be configured with a custom `ServerCodecConfigurer`, `RequestedContentTypeResolver` and even a `ReactiveAdapterRegistry`. -The latter provides a mechanism where we can return a reply as any reactive type - Reactor `Flux`, RxJava `Observable`, `Flowable` etc. -This way, we can simply implement https://en.wikipedia.org/wiki/Server-sent_events[Server Sent Events] scenarios with Spring Integration components: - -[source,java] ----- -@Bean -public IntegrationFlow sseFlow() { - return IntegrationFlows - .from(Http.inboundReactiveGateway("/sse") - .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) - .handle((p, h) -> Flux.just("foo", "bar", "baz")) - .get(); -} ----- - [[http-outbound]] === Http Outbound Components ==== HttpRequestExecutingMessageHandler @@ -244,43 +192,6 @@ The `expected-response-type` must be compatible with the (configured or default) Of course, this can be an abstract class, or even an interface (such as `java.io.Serializable` when using java serialization and `Content-Type: application/x-java-serialized-object`). ===== -==== ReactiveHttpRequestExecutingMessageHandler - -The `ReactiveHttpRequestExecutingMessageHandler` (starting with _version 5.0_) implementation is very similar to `HttpRequestExecutingMessageHandler`, using a `WebClient` from the Spring Framework WebFlux module. -To configure it, define a bean like this: - -[source,xml] ----- - - - - ----- - -You can configure a `WebClient` instance to use: - -[source,xml] ----- - - - - - - - ----- - -The `WebClient` `exchange()` operation returns a `Mono` which is mapped to the `AbstractIntegrationMessageBuilder` reactive support (using `Mono.map()`) as the output from the `ReactiveHttpRequestExecutingMessageHandler`. -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 `ReactiveHttpRequestExecutingMessageHandler`. - -See http://docs.spring.io/spring/docs/5.0.0.M5/spring-framework-reference/html/web-reactive.html#web-reactive-client[WebFlux documentation] and https://projectreactor.io/[Project Reactor] for more information. - -For other settings like cookie, uri variables, etc, please see `HttpRequestExecutingMessageHandler` above. - [[http-namespace]] === HTTP Namespace Support @@ -327,6 +238,7 @@ To process requests that do expect a response, use an _inbound-gateway_: reply-channel="responses"/> ---- +[[http-request-mapping]] ==== Request Mapping Support NOTE: _Spring Integration 3.0_ is improving the REST support by introducing the http://static.springsource.org/spring-integration/api/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.html[IntegrationRequestMappingHandlerMapping]. @@ -570,7 +482,7 @@ Alternatively, you may wish to consider using JSON instead which is enabled by s ===== Beginning with Spring Integration 2.2 you can also determine the HTTP Method dynamically using SpEL and the _http-method-expression_ attribute. -Note that this attribute is obviously murually exclusive with _http-method_ You can also use `expected-response-type-expression` attribute instead of `expected-response-type` and provide any valid SpEL expression that determines the type of the response. +Note that this attribute is obviously mutually exclusive with _http-method_ You can also use `expected-response-type-expression` attribute instead of `expected-response-type` and provide any valid SpEL expression that determines the type of the response. [source,xml] ---- ---- -If you want to execute the http request in a reactive, non-blocking way, you can use the `outbound-reactive-gateway` or `outbound-reactive-channel-adapter`. - -[source,xml] ----- - - - - ----- - [NOTE] ===== To specify the URL; you can use either the 'url' attribute or the 'url-expression' attribute. @@ -739,26 +625,6 @@ public RequestMapping mapping() { requestMapping.setMethods(HttpMethod.POST); return requestMapping; } - -@Bean -public ReactiveHttpInboundEndpoint jsonInboundEndpoint() { - ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); - RequestMapping requestMapping = new RequestMapping(); - requestMapping.setPathPatterns("/persons"); - endpoint.setRequestMapping(requestMapping); - endpoint.setRequestChannel(fluxResultChannel()); - return endpoint; -} - -@Bean -public MessageChannel fluxResultChannel() { - return new FluxMessageChannel(); -} - -@ServiceActivator(inputChannel = "fluxResultChannel") -Flux getPersons() { - return Flux.just(new Person("Jane"), new Person("Jason"), new Person("John")); -} ---- .Inbound Gateway Using the Java DSL @@ -772,17 +638,6 @@ public IntegrationFlow inbound() { .channel("httpRequest") .get(); } - -@Bean -public IntegrationFlow httpReactiveInboundChannelAdapterFlow() { - return IntegrationFlows - .from(Http.inboundReactiveChannelAdapter("/reactivePost") - .requestMapping(m -> m.methods(HttpMethod.POST)) - .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class)) - .statusCodeFunction(m -> HttpStatus.ACCEPTED)) - .channel(c -> c.queue("storeChannel")) - .get(); -} ---- .Outbound Gateway Using Java Configuration @@ -797,16 +652,6 @@ public HttpRequestExecutingMessageHandler outbound() { handler.setExpectedResponseType(String.class); return handler; } - -@ServiceActivator(inputChannel = "reactiveHttpOutRequest") -@Bean -public ReactiveHttpRequestExecutingMessageHandler reactiveOutbound(WebClient client) { - ReactiveHttpRequestExecutingMessageHandler handler = - new ReactiveHttpRequestExecutingMessageHandler("http://localhost:8080/foo", client); - handler.setHttpMethod(HttpMethod.POST); - handler.setExpectedResponseType(String.class); - return handler; -} ---- .Outbound Gateway Using the Java DSL @@ -820,18 +665,6 @@ public IntegrationFlow outbound() { .expectedResponseType(String.class)) .get(); } - -@Bean -public IntegrationFlow outboundReactive() { - return f -> f - .handle(Http.>outboundReactiveGateway(m -> - UriComponentsBuilder.fromUriString("http://localhost:8080/foo") - .queryParams(m.getPayload()) - .build() - .toUri()) - .httpMethod(HttpMethod.GET) - .expectedResponseType(String.class)); -} ---- [[http-timeout]] diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index 3ded095582..13c402e74d 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -98,6 +98,8 @@ include::./ip.adoc[] include::./twitter.adoc[] +include::./webflux.adoc[] + include::./web-sockets.adoc[] include::./ws.adoc[] diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc new file mode 100644 index 0000000000..2ee3a055e2 --- /dev/null +++ b/src/reference/asciidoc/webflux.adoc @@ -0,0 +1,240 @@ +[[webflux]] +== WebFlux Support + +[[webflux-intro]] +=== Introduction + +The WebFlux Spring Integration module (`spring-integration-webflux`) allows for the execution of HTTP requests and the processing of inbound HTTP requests in Reactive manner. +The WebFlux support consists of the following gateway implementations: `WebFluxInboundEndpoint`, `WebFluxRequestExecutingMessageHandler`. +The implementation is fully based on the Spring http://docs.spring.io/spring/docs/5.0.0.RC3/spring-framework-reference/web.html#web-reactive[WebFlux] and https://projectreactor.io/[Project Reactor] foundations. +Also see <> for more information since many options are shared between reactive and regular HTTP components. + +[[webflux-inbound]] +=== WebFlux Inbound Components + +Starting with _version 5.0_, the `WebFluxInboundEndpoint`, `WebHandler`, implementation is provided. +This component is similar to the MVC-based `HttpRequestHandlingEndpointSupport` with which it shares some common options via the newly extracted `BaseHttpInboundEndpoint`. +Instead of MVC, it is used in the Spring WebFlux Reactive environment. +A simple sample for explanation: + +[source,java] +---- +@Configuration +@EnableWebFlux +@EnableIntegration +public class ReactiveHttpConfiguration { + + @Bean + public WebFluxInboundEndpoint simpleInboundEndpoint() { + WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/test"); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannelName("serviceChannel"); + return endpoint; + } + + @ServiceActivator(inputChannel = "serviceChannel") + String service() { + return "It works!"; + } + +} +---- + +As can be seen, the configuration is similar to the `HttpRequestHandlingEndpointSupport` mentioned above, except that we use `@EnableWebFlux` to add the WebFlux infrastructure to our integration application. +Also, the `WebFluxInboundEndpoint` performs `sendAndReceive` operation to the downstream flow using back-pressure, on demand based capabilities, provided by the reactive HTTP server implementation. + +NOTE: The reply part is non-blocking as well and based on the internal `FutureReplyChannel` which is flat-mapped to a reply `Mono` for on demand resolution. + +The `WebFluxInboundEndpoint` can be configured with a custom `ServerCodecConfigurer`, `RequestedContentTypeResolver` and even a `ReactiveAdapterRegistry`. +The latter provides a mechanism where we can return a reply as any reactive type - Reactor `Flux`, RxJava `Observable`, `Flowable` etc. +This way, we can simply implement https://en.wikipedia.org/wiki/Server-sent_events[Server Sent Events] scenarios with Spring Integration components: + +[source,java] +---- +@Bean +public IntegrationFlow sseFlow() { + return IntegrationFlows + .from(WebFlux.inboundGateway("/sse") + .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) + .handle((p, h) -> Flux.just("foo", "bar", "baz")) + .get(); +} +---- + +Also see <> and <> for more possible configuration options. + +[[webflux-outbound]] +=== WebFlux Outbound Components + +The `WebFluxRequestExecutingMessageHandler` (starting with _version 5.0_) implementation is very similar to `HttpRequestExecutingMessageHandler`, using a `WebClient` from the Spring Framework WebFlux module. +To configure it, define a bean like this: + +[source,xml] +---- + + + + +---- + +You can configure a `WebClient` instance to use: + +[source,xml] +---- + + + + + + + +---- + +The `WebClient` `exchange()` operation returns a `Mono` which is mapped to the `AbstractIntegrationMessageBuilder` reactive support (using `Mono.map()`) 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`. + + +Also see <> for more possible configuration options. + +[[webflux-namespace]] +=== WebFlux Namespace Support + +==== Introduction + +Spring Integration provides a _webflux_ namespace and the corresponding schema definition. +To include it in your configuration, simply provide the following namespace declaration in your application context configuration file: + +[source,xml] +---- + + + ... + +---- + +==== Inbound + + +==== Outbound + +If you want to execute the http request in a reactive, non-blocking way, you can use the `outbound-gateway` or `outbound-channel-adapter`. + +[source,xml] +---- + + + + +---- + + +[[webflux-java-config]] +=== Configuring WebFlux Endpoints with Java + +.Inbound Gateway Using Java Configuration +[source, java] +---- +@Bean +public WebFluxInboundEndpoint jsonInboundEndpoint() { + WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/persons"); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannel(fluxResultChannel()); + return endpoint; +} + +@Bean +public MessageChannel fluxResultChannel() { + return new FluxMessageChannel(); +} + +@ServiceActivator(inputChannel = "fluxResultChannel") +Flux getPersons() { + return Flux.just(new Person("Jane"), new Person("Jason"), new Person("John")); +} +---- + +.Inbound Gateway Using the Java DSL +[source, java] +---- +@Bean +public IntegrationFlow inboundChannelAdapterFlow() { + return IntegrationFlows + .from(WebFlux.inboundChannelAdapter("/reactivePost") + .requestMapping(m -> m.methods(HttpMethod.POST)) + .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class)) + .statusCodeFunction(m -> HttpStatus.ACCEPTED)) + .channel(c -> c.queue("storeChannel")) + .get(); +} +---- + +.Outbound Gateway Using Java Configuration +[source, java] +---- +@ServiceActivator(inputChannel = "reactiveHttpOutRequest") +@Bean +public WebFluxRequestExecutingMessageHandler reactiveOutbound(WebClient client) { + WebFluxRequestExecutingMessageHandler handler = + new WebFluxRequestExecutingMessageHandler("http://localhost:8080/foo", client); + handler.setHttpMethod(HttpMethod.POST); + handler.setExpectedResponseType(String.class); + return handler; +} +---- + +.Outbound Gateway Using the Java DSL +[source, java] +---- +@Bean +public IntegrationFlow outboundReactive() { + return f -> f + .handle(WebFlux.>outboundGateway(m -> + UriComponentsBuilder.fromUriString("http://localhost:8080/foo") + .queryParams(m.getPayload()) + .build() + .toUri()) + .httpMethod(HttpMethod.GET) + .expectedResponseType(String.class)); +} +---- + + +[[webflux-header-mapping]] +=== 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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 5f5d5c8835..07aab180e9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -29,11 +29,11 @@ The new `MongoDbOutboundGateway` allows you to make queries to the database on d See <> for more information. -==== HTTP Reactive Inbound and Outbound Gateways and Channel Adapters +==== WebFlux Gateways and Channel Adapters -The new `ReactiveHttpInboundEndpoint` and `ReactiveHttpRequestExecutingMessageHandler` add support for Spring WebFlux Framework gateways and channel adapters. +The new WebFlux support module has been introduced for Spring WebFlux Framework gateways and channel adapters. -See <> for more information. +See <> for more information. ==== Content Type Conversion