From 19c5402079e3f8e35bbe7db4ab1aa9b134db17a1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 4 Apr 2017 15:27:29 -0400 Subject: [PATCH] INT-4208: Add WebFlux-Based HttpMessageHandler JIRA: https://jira.spring.io/browse/INT-4208 Since `AsyncRestTemplate` is deprecated in Spring 5.0, it doesn't make sense to promote that feature via our new `AsyncHttpRequestExecutingMessageHandler` component * Rework (and rename) `AsyncHttpRequestExecutingMessageHandler` to `ReactiveHttpRequestExecutingMessageHandler` and make it based on the WebFlux `WebClient` * Fix Java DSL (`ReactiveHttpMessageHandlerSpec`) and all tests according a new logic in the `ReactiveHttpRequestExecutingMessageHandler` * Fix XML namespace support to use new `ReactiveHttpRequestExecutingMessageHandler` and don't expose unused options like `converters` and `request-factory` * Fix `What's New` and `http.adoc` * To remain with `async` mode for the `ReactiveHttpRequestExecutingMessageHandler` behavior support fix `AbstractMessageProducingHandler` to adapt reply `Mono` to the `SettableListenableFuture` * Introduce new `reactive` behavior for the `AbstractMessageProducingHandler` when `outputChannel` is `ReactiveSubscribableChannel` and perform `.subscribeTo(Publisher)` for `Publisher` reply * Upgrade to Spring AMQP `2.0 M3` * Downgrade to Spring Security `4.2.2` Address PR comments Add async error handling to the one-way case Minor polishing and checkstyle fix --- build.gradle | 6 +- .../AbstractMessageProducingHandler.java | 80 ++++--- .../http/config/HttpAdapterParsingUtils.java | 50 ++-- .../http/config/HttpNamespaceHandler.java | 4 +- .../HttpOutboundChannelAdapterParser.java | 26 +-- .../config/HttpOutboundGatewayParser.java | 37 ++- .../http/dsl/AsyncHttpMessageHandlerSpec.java | 69 ------ .../http/dsl/BaseHttpMessageHandlerSpec.java | 30 +-- .../integration/http/dsl/Http.java | 220 +++++++++--------- .../http/dsl/HttpMessageHandlerSpec.java | 30 ++- .../dsl/ReactiveHttpMessageHandlerSpec.java | 60 +++++ ...actHttpRequestExecutingMessageHandler.java | 51 ++-- ...yncHttpRequestExecutingMessageHandler.java | 162 ------------- .../HttpRequestExecutingMessageHandler.java | 16 +- ...iveHttpRequestExecutingMessageHandler.java | 198 ++++++++++++++++ .../config/spring-integration-http-5.0.xsd | 110 ++++----- ...boundChannelAdapterParserTests-context.xml | 63 ++--- ...HttpOutboundChannelAdapterParserTests.java | 83 +++---- ...HttpOutboundGatewayParserTests-context.xml | 86 +++---- .../HttpOutboundGatewayParserTests.java | 80 +++---- .../integration/http/dsl/HttpDslTests.java | 48 ++-- ...tpRequestExecutingMessageHandlerTests.java | 63 ----- ...tpRequestExecutingMessageHandlerTests.java | 136 +++++++++++ src/reference/asciidoc/http.adoc | 38 +-- src/reference/asciidoc/whats-new.adoc | 6 +- 25 files changed, 939 insertions(+), 813 deletions(-) delete mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/dsl/AsyncHttpMessageHandlerSpec.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java delete mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandler.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java delete mode 100644 spring-integration-http/src/test/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandlerTests.java create mode 100644 spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java diff --git a/build.gradle b/build.gradle index e17ca5e857..9bbc55cb2e 100644 --- a/build.gradle +++ b/build.gradle @@ -130,12 +130,12 @@ subprojects { subproject -> servletApiVersion = '3.1.0' slf4jVersion = "1.7.21" smackVersion = '4.1.7' - springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.0.0.BUILD-SNAPSHOT' + springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.0.0.M3' springDataJpaVersion = '2.0.0.M2' springDataMongoVersion = '2.0.0.M2' springDataRedisVersion = '2.0.0.M2' springGemfireVersion = '2.0.0.M2' - springSecurityVersion = '5.0.0.BUILD-SNAPSHOT' + springSecurityVersion = '4.2.2.RELEASE' springSocialTwitterVersion = '2.0.0.M1' springRetryVersion = '1.2.0.RELEASE' springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.0.M5' @@ -389,8 +389,10 @@ project('spring-integration-http') { dependencies { compile project(":spring-integration-core") compile "org.springframework:spring-webmvc:$springVersion" + compile "org.springframework:spring-webflux:$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" diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 4a98befd39..8f2466ad9e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -21,7 +21,10 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import org.reactivestreams.Publisher; + import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.ReactiveSubscribableChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; @@ -37,6 +40,10 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; +import org.springframework.util.concurrent.SettableListenableFuture; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; /** * The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}. @@ -44,6 +51,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author David Liu * @author Artem Bilan * @author Gary Russell + * * since 4.1 */ public abstract class AbstractMessageProducingHandler extends AbstractMessageHandler @@ -181,36 +189,56 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } } - if (this.async && reply instanceof ListenableFuture) { - ListenableFuture future = (ListenableFuture) reply; - final Object theReplyChannel = replyChannel; - future.addCallback(new ListenableFutureCallback() { + if (this.async && (reply instanceof ListenableFuture || reply instanceof Publisher)) { + if (reply instanceof ListenableFuture || !(getOutputChannel() instanceof ReactiveSubscribableChannel)) { + ListenableFuture future; + if (reply instanceof ListenableFuture) { + future = (ListenableFuture) reply; + } + else { + SettableListenableFuture settableListenableFuture = new SettableListenableFuture<>(); - @Override - public void onSuccess(Object result) { - Message replyMessage = null; - try { - replyMessage = createOutputMessage(result, requestHeaders); - sendOutput(replyMessage, theReplyChannel, false); - } - catch (Exception e) { - Exception exceptionToLogAndSend = e; - if (!(e instanceof MessagingException)) { - exceptionToLogAndSend = new MessageHandlingException(requestMessage, e); - if (replyMessage != null) { - exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); - } + Mono.from((Publisher) reply) + .subscribe(settableListenableFuture::set, settableListenableFuture::setException); + + future = settableListenableFuture; + } + + Object theReplyChannel = replyChannel; + future.addCallback(new ListenableFutureCallback() { + + @Override + public void onSuccess(Object result) { + Message replyMessage = null; + try { + replyMessage = createOutputMessage(result, requestHeaders); + sendOutput(replyMessage, theReplyChannel, false); + } + catch (Exception e) { + Exception exceptionToLogAndSend = e; + if (!(e instanceof MessagingException)) { + exceptionToLogAndSend = new MessageHandlingException(requestMessage, e); + if (replyMessage != null) { + exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); + } + } + logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend); + onFailure(exceptionToLogAndSend); } - logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend); - onFailure(exceptionToLogAndSend); } - } - @Override - public void onFailure(Throwable ex) { - sendErrorMessage(requestMessage, ex); - } - }); + @Override + public void onFailure(Throwable ex) { + sendErrorMessage(requestMessage, ex); + } + + }); + } + else { + ((ReactiveSubscribableChannel) getOutputChannel()) + .subscribeTo(Flux.from((Publisher) reply) + .map(result -> createOutputMessage(result, requestHeaders))); + } } else { sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false); diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpAdapterParsingUtils.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpAdapterParsingUtils.java index d8c9a8ec0f..cf47f45e01 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpAdapterParsingUtils.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpAdapterParsingUtils.java @@ -41,34 +41,21 @@ import org.springframework.util.xml.DomUtils; abstract class HttpAdapterParsingUtils { static final String[] SYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES = { - "request-factory", "error-handler", "message-converters" - }; - - static final String[] ASYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES = { - "async-request-factory", "error-handler", "message-converters" + "request-factory", "error-handler", "message-converters" }; static void verifyNoRestTemplateAttributes(Element element, ParserContext parserContext) { for (String attributeName : SYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { if (element.hasAttribute(attributeName)) { parserContext.getReaderContext().error("When providing a 'rest-template' reference, the '" - + attributeName + "' attribute is not allowed.", - parserContext.extractSource(element)); - } - } - } - - static void verifyNoAsyncRestTemplateAttributes(Element element, ParserContext parserContext) { - for (String attributeName : ASYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { - if (element.hasAttribute(attributeName)) { - parserContext.getReaderContext().error("When providing an 'async-rest-template' reference, the '" + attributeName + "' attribute is not allowed.", parserContext.extractSource(element)); } } } - static void configureUriVariableExpressions(BeanDefinitionBuilder builder, ParserContext parserContext, Element element) { + static void configureUriVariableExpressions(BeanDefinitionBuilder builder, ParserContext parserContext, + Element element) { String uriVariablesExpression = element.getAttribute("uri-variables-expression"); List uriVariableElements = DomUtils.getChildElementsByTagName(element, "uri-variable"); @@ -79,8 +66,9 @@ abstract class HttpAdapterParsingUtils { parserContext.getReaderContext().error("'uri-variables-expression' attribute " + "and 'uri-variable' sub-elements are mutually exclusive.", element); } - BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class) - .addConstructorArgValue(uriVariablesExpression); + BeanDefinitionBuilder beanDefinitionBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class) + .addConstructorArgValue(uriVariablesExpression); builder.addPropertyValue("uriVariablesExpression", beanDefinitionBuilder.getBeanDefinition()); } @@ -89,21 +77,24 @@ abstract class HttpAdapterParsingUtils { for (Element uriVariableElement : uriVariableElements) { String name = uriVariableElement.getAttribute("name"); String expression = uriVariableElement.getAttribute("expression"); - BeanDefinitionBuilder factoryBeanBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); + BeanDefinitionBuilder factoryBeanBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); factoryBeanBuilder.addConstructorArgValue(expression); - uriVariableExpressions.put(name, factoryBeanBuilder.getBeanDefinition()); + uriVariableExpressions.put(name, factoryBeanBuilder.getBeanDefinition()); } builder.addPropertyValue("uriVariableExpressions", uriVariableExpressions); } } - static void configureUrlConstructorArg(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + static void configureUrlConstructorArg(Element element, ParserContext parserContext, + BeanDefinitionBuilder builder) { String urlAttribute = element.getAttribute("url"); String urlExpressionAttribute = element.getAttribute("url-expression"); boolean hasUrlAttribute = StringUtils.hasText(urlAttribute); boolean hasUrlExpressionAttribute = StringUtils.hasText(urlExpressionAttribute); - if (!(hasUrlAttribute ^ hasUrlExpressionAttribute)) { - parserContext.getReaderContext().error("Adapter must have exactly one of 'url' or 'url-expression'", element); + if (hasUrlAttribute == hasUrlExpressionAttribute) { + parserContext.getReaderContext() + .error("Adapter must have exactly one of 'url' or 'url-expression'", element); } RootBeanDefinition expressionDef; if (hasUrlAttribute) { @@ -126,8 +117,9 @@ abstract class HttpAdapterParsingUtils { boolean hasHttpMethodExpression = StringUtils.hasText(httpMethodExpression); if (hasHttpMethod && hasHttpMethodExpression) { - parserContext.getReaderContext().error("The 'http-method' and 'http-method-expression' are mutually exclusive. " + - "You can only have one or the other", element); + parserContext.getReaderContext() + .error("The 'http-method' and 'http-method-expression' are mutually exclusive. " + + "You can only have one or the other", element); } RootBeanDefinition expressionDef = null; @@ -144,7 +136,8 @@ abstract class HttpAdapterParsingUtils { } } - static void setExpectedResponseOrExpression(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + static void setExpectedResponseOrExpression(Element element, ParserContext parserContext, + BeanDefinitionBuilder builder) { String expectedResponseType = element.getAttribute("expected-response-type"); String expectedResponseTypeExpression = element.getAttribute("expected-response-type-expression"); @@ -152,8 +145,9 @@ abstract class HttpAdapterParsingUtils { boolean hasExpectedResponseTypeExpression = StringUtils.hasText(expectedResponseTypeExpression); if (hasExpectedResponseType && hasExpectedResponseTypeExpression) { - parserContext.getReaderContext().error("The 'expected-response-type' and 'expected-response-type-expression' are mutually exclusive. " + - "You can only have one or the other", element); + parserContext.getReaderContext() + .error("The 'expected-response-type' and 'expected-response-type-expression' are mutually exclusive. " + + "You can only have one or the other", element); } RootBeanDefinition expressionDef = null; 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 31957121c6..93ceb5d3f4 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,8 @@ public class HttpNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("inbound-gateway", new HttpInboundEndpointParser(true)); registerBeanDefinitionParser("outbound-channel-adapter", new HttpOutboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-gateway", new HttpOutboundGatewayParser()); - registerBeanDefinitionParser("outbound-async-channel-adapter", new HttpOutboundChannelAdapterParser()); - registerBeanDefinitionParser("outbound-async-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 8230098bd8..7dfad2ec0c 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 @@ -23,8 +23,8 @@ 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.AsyncHttpRequestExecutingMessageHandler; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; +import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; import org.springframework.util.StringUtils; /** @@ -35,6 +35,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan * @author Shiliang Li + * * @since 2.0 */ public class HttpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { @@ -42,9 +43,9 @@ public class HttpOutboundChannelAdapterParser extends AbstractOutboundChannelAda @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder; - boolean async = element.getLocalName().contains("async"); - if (async) { - builder = BeanDefinitionBuilder.genericBeanDefinition(AsyncHttpRequestExecutingMessageHandler.class); + boolean reactive = element.getLocalName().contains("reactive"); + if (reactive) { + builder = BeanDefinitionBuilder.genericBeanDefinition(ReactiveHttpRequestExecutingMessageHandler.class); } else { builder = BeanDefinitionBuilder.genericBeanDefinition(HttpRequestExecutingMessageHandler.class); @@ -55,17 +56,11 @@ public class HttpOutboundChannelAdapterParser extends AbstractOutboundChannelAda IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encode-uri"); HttpAdapterParsingUtils.setHttpMethodOrExpression(element, parserContext, builder); - if (async) { - String asyncTemplateRef = element.getAttribute("async-rest-template"); + if (reactive) { + String webClientRef = element.getAttribute("web-client"); - if (StringUtils.hasText(asyncTemplateRef)) { - HttpAdapterParsingUtils.verifyNoAsyncRestTemplateAttributes(element, parserContext); - builder.addConstructorArgReference(asyncTemplateRef); - } - else { - for (String referenceAttributeName : HttpAdapterParsingUtils.ASYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, referenceAttributeName); - } + if (StringUtils.hasText(webClientRef)) { + builder.addConstructorArgReference(webClientRef); } } else { @@ -95,7 +90,8 @@ public class HttpOutboundChannelAdapterParser extends AbstractOutboundChannelAda else if (StringUtils.hasText(mappedRequestHeaders)) { BeanDefinitionBuilder headerMapperBuilder = BeanDefinitionBuilder.genericBeanDefinition( "org.springframework.integration.http.support.DefaultHttpHeaderMapper"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(headerMapperBuilder, element, "mapped-request-headers", "outboundHeaderNames"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(headerMapperBuilder, element, + "mapped-request-headers", "outboundHeaderNames"); builder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition()); } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset"); 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 b75f3ff6e5..f23e4c1148 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 @@ -22,8 +22,8 @@ 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.AsyncHttpRequestExecutingMessageHandler; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; +import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; import org.springframework.util.StringUtils; /** @@ -45,9 +45,9 @@ public class HttpOutboundGatewayParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder; - boolean async = element.getLocalName().contains("async"); - if (async) { - builder = BeanDefinitionBuilder.genericBeanDefinition(AsyncHttpRequestExecutingMessageHandler.class); + boolean reactive = element.getLocalName().contains("reactive"); + if (reactive) { + builder = BeanDefinitionBuilder.genericBeanDefinition(ReactiveHttpRequestExecutingMessageHandler.class); } else { builder = BeanDefinitionBuilder.genericBeanDefinition(HttpRequestExecutingMessageHandler.class); @@ -57,16 +57,10 @@ public class HttpOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encode-uri"); HttpAdapterParsingUtils.setHttpMethodOrExpression(element, parserContext, builder); - if (async) { - String asyncTemplateRef = element.getAttribute("async-rest-template"); - if (StringUtils.hasText(asyncTemplateRef)) { - HttpAdapterParsingUtils.verifyNoAsyncRestTemplateAttributes(element, parserContext); - builder.addConstructorArgReference(asyncTemplateRef); - } - else { - for (String referenceAttributeName : HttpAdapterParsingUtils.ASYNC_REST_TEMPLATE_REFERENCE_ATTRIBUTES) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, referenceAttributeName); - } + if (reactive) { + String webClientRef = element.getAttribute("web-client"); + if (StringUtils.hasText(webClientRef)) { + builder.addConstructorArgReference(webClientRef); } } else { @@ -88,8 +82,10 @@ public class HttpOutboundGatewayParser extends AbstractConsumerEndpointParser { String mappedResponseHeaders = element.getAttribute("mapped-response-headers"); if (StringUtils.hasText(headerMapper)) { if (StringUtils.hasText(mappedRequestHeaders) || StringUtils.hasText(mappedResponseHeaders)) { - parserContext.getReaderContext().error("Neither 'mapped-request-headers' or 'mapped-response-headers' " + - "attributes are allowed when a 'header-mapper' has been specified.", parserContext.extractSource(element)); + parserContext.getReaderContext() + .error("Neither 'mapped-request-headers' or 'mapped-response-headers' " + + "attributes are allowed when a 'header-mapper' has been specified.", + parserContext.extractSource(element)); return null; } builder.addPropertyReference("headerMapper", headerMapper); @@ -98,12 +94,15 @@ public class HttpOutboundGatewayParser extends AbstractConsumerEndpointParser { BeanDefinitionBuilder headerMapperBuilder = BeanDefinitionBuilder.genericBeanDefinition( "org.springframework.integration.http.support.DefaultHttpHeaderMapper"); headerMapperBuilder.setFactoryMethod("outboundMapper"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(headerMapperBuilder, element, "mapped-request-headers", "outboundHeaderNames"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(headerMapperBuilder, element, "mapped-response-headers", "inboundHeaderNames"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(headerMapperBuilder, element, + "mapped-request-headers", "outboundHeaderNames"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(headerMapperBuilder, element, + "mapped-response-headers", "inboundHeaderNames"); builder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition()); } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-request-payload", "extractPayload"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-request-payload", + "extractPayload"); HttpAdapterParsingUtils.setExpectedResponseOrExpression(element, parserContext, builder); diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/AsyncHttpMessageHandlerSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/AsyncHttpMessageHandlerSpec.java deleted file mode 100644 index 9a819a8742..0000000000 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/AsyncHttpMessageHandlerSpec.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * 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.http.dsl; - -import java.net.URI; - -import org.springframework.expression.Expression; -import org.springframework.expression.common.LiteralExpression; -import org.springframework.http.client.AsyncClientHttpRequestFactory; -import org.springframework.integration.expression.ValueExpression; -import org.springframework.integration.http.outbound.AsyncHttpRequestExecutingMessageHandler; -import org.springframework.util.Assert; -import org.springframework.web.client.AsyncRestTemplate; - -/** - * The {@link BaseHttpMessageHandlerSpec} implementation for the {@link AsyncHttpRequestExecutingMessageHandler}. - * @author Shiliang Li - * @since 5.0 - * @see AsyncHttpRequestExecutingMessageHandler - */ -public class AsyncHttpMessageHandlerSpec - extends BaseHttpMessageHandlerSpec { - - private final AsyncRestTemplate asyncRestTemplate; - - AsyncHttpMessageHandlerSpec(URI uri, AsyncRestTemplate asyncRestTemplate) { - this(new ValueExpression<>(uri), asyncRestTemplate); - } - - AsyncHttpMessageHandlerSpec(String uri, AsyncRestTemplate asyncRestTemplate) { - this(new LiteralExpression(uri), asyncRestTemplate); - } - - AsyncHttpMessageHandlerSpec(Expression uriExpression, AsyncRestTemplate asyncRestTemplate) { - super(new AsyncHttpRequestExecutingMessageHandler(uriExpression, asyncRestTemplate)); - this.asyncRestTemplate = asyncRestTemplate; - } - - /** - * Set the {@link AsyncClientHttpRequestFactory} for the underlying {@link AsyncRestTemplate}. - * @param asyncRequestFactory The request factory. - * @return the spec - */ - public AsyncHttpMessageHandlerSpec asyncRequestFactory(AsyncClientHttpRequestFactory asyncRequestFactory) { - Assert.isNull(this.asyncRestTemplate, - "the 'requestFactory' must be specified on the provided 'restTemplate': " + this.asyncRestTemplate); - this.target.setAsyncRequestFactory(asyncRequestFactory); - return this; - } - - @Override - protected boolean isRestTemplateSet() { - return this.asyncRestTemplate != null; - } -} 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 4e8028742e..bbde39d722 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 @@ -16,7 +16,6 @@ package org.springframework.integration.http.dsl; -import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -27,7 +26,6 @@ import org.springframework.core.ParameterizedTypeReference; import org.springframework.expression.Expression; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; -import org.springframework.http.converter.HttpMessageConverter; import org.springframework.integration.dsl.ComponentsRegistration; import org.springframework.integration.dsl.MessageHandlerSpec; import org.springframework.integration.expression.FunctionExpression; @@ -37,8 +35,6 @@ import org.springframework.integration.http.support.DefaultHttpHeaderMapper; import org.springframework.integration.mapping.HeaderMapper; import org.springframework.messaging.Message; import org.springframework.util.Assert; -import org.springframework.web.client.ResponseErrorHandler; -import org.springframework.web.client.RestTemplate; /** * The base {@link MessageHandlerSpec} for {@link AbstractHttpRequestExecutingMessageHandler}s. @@ -319,29 +315,5 @@ public abstract class BaseHttpMessageHandlerSpec... messageConverters) { - Assert.isTrue(!isRestTemplateSet(), "the 'messageConverters' must be specified on the provided restTemplate"); - this.target.setMessageConverters(Arrays.asList(messageConverters)); - return _this(); - } - - protected abstract boolean isRestTemplateSet(); + protected abstract boolean isClientSet(); } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/Http.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/Http.java index 1725c5e92e..14ebba59cd 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/Http.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/Http.java @@ -27,8 +27,8 @@ import org.springframework.integration.http.inbound.HttpRequestHandlingMessaging import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import org.springframework.web.client.AsyncRestTemplate; import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; /** * The HTTP components Factory. @@ -128,90 +128,95 @@ public final class Http { } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for one-way adapter based on provided {@link URI}. + * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter based on provided {@link URI}. * @param uri the {@link URI} to send requests. - * @return the AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static AsyncHttpMessageHandlerSpec outboundAsyncChannelAdapter(URI uri) { - return outboundAsyncChannelAdapter(uri, null); + public static ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(URI uri) { + return outboundReactiveChannelAdapter(uri, null); } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for one-way adapter based on provided {@code uri}. + * Create an {@link ReactiveHttpMessageHandlerSpec} builder for one-way adapter based on provided {@code uri}. * @param uri the {@code uri} to send requests. - * @return the AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static AsyncHttpMessageHandlerSpec outboundAsyncChannelAdapter(String uri) { - return outboundAsyncChannelAdapter(uri, null); + public static ReactiveHttpMessageHandlerSpec outboundReactiveChannelAdapter(String uri) { + return outboundReactiveChannelAdapter(uri, null); } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for one-way adapter based on provided {@code Function} + * Create an {@link ReactiveHttpMessageHandlerSpec} 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 AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static

AsyncHttpMessageHandlerSpec outboundAsyncChannelAdapter(Function, ?> uriFunction) { - return outboundAsyncChannelAdapter(new FunctionExpression<>(uriFunction)); + public static

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

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

AsyncHttpMessageHandlerSpec outboundAsyncChannelAdapter(Function, ?> uriFunction, - AsyncRestTemplate asyncRestTemplate) { - return outboundAsyncChannelAdapter(new FunctionExpression<>(uriFunction), asyncRestTemplate); - } - - /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for one-way adapter + * 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 AsyncRestTemplate} for HTTP exchanges. + * against request message. * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. - * @param asyncRestTemplate {@link AsyncRestTemplate} to use. - * @return the AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static AsyncHttpMessageHandlerSpec outboundAsyncChannelAdapter(Expression uriExpression, AsyncRestTemplate asyncRestTemplate) { - return new AsyncHttpMessageHandlerSpec(uriExpression, asyncRestTemplate).expectReply(false); + 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); } /** @@ -302,90 +307,93 @@ public final class Http { } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for request-reply gateway based on provided {@link URI}. + * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway + * based on provided {@link URI}. * @param uri the {@link URI} to send requests. - * @return the AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static AsyncHttpMessageHandlerSpec outboundAsyncGateway(URI uri) { - return outboundAsyncGateway(uri, null); + public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(URI uri) { + return outboundReactiveGateway(uri, null); } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for request-reply gateway based on provided {@code uri}. + * Create an {@link ReactiveHttpMessageHandlerSpec} builder for request-reply gateway + * based on provided {@code uri}. * @param uri the {@code uri} to send requests. - * @return the AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static AsyncHttpMessageHandlerSpec outboundAsyncGateway(String uri) { - return outboundAsyncGateway(uri, null); + public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(String uri) { + return outboundReactiveGateway(uri, null); } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for request-reply gateway + * 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 AsyncHttpMessageHandlerSpec instance + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static

AsyncHttpMessageHandlerSpec outboundAsyncGateway(Function, ?> uriFunction) { - return outboundAsyncGateway(new FunctionExpression<>(uriFunction)); + public static

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

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

AsyncHttpMessageHandlerSpec outboundAsyncGateway(Function, ?> uriFunction, - AsyncRestTemplate asyncRestTemplate) { - return outboundAsyncGateway(new FunctionExpression<>(uriFunction), asyncRestTemplate); + public static

ReactiveHttpMessageHandlerSpec outboundReactiveGateway(Function, ?> uriFunction, + WebClient webClient) { + return outboundReactiveGateway(new FunctionExpression<>(uriFunction), webClient); } /** - * Create an {@link AsyncHttpMessageHandlerSpec} builder for request-reply gateway + * 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 AsyncRestTemplate} for HTTP exchanges. + * against request message and {@link WebClient} for HTTP exchanges. * @param uriExpression the SpEL {@link Expression} to evaluate {@code uri} at runtime. - * @param asyncRestTemplate {@link AsyncRestTemplate} to use. - * @return the AsyncHttpMessageHandlerSpec instance + * @param webClient {@link WebClient} to use. + * @return the ReactiveHttpMessageHandlerSpec instance */ - public static AsyncHttpMessageHandlerSpec outboundAsyncGateway(Expression uriExpression, AsyncRestTemplate asyncRestTemplate) { - return new AsyncHttpMessageHandlerSpec(uriExpression, asyncRestTemplate); + public static ReactiveHttpMessageHandlerSpec outboundReactiveGateway(Expression uriExpression, + WebClient webClient) { + return new ReactiveHttpMessageHandlerSpec(uriExpression, webClient); } /** diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpMessageHandlerSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpMessageHandlerSpec.java index fe910425dd..fae5af2b55 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpMessageHandlerSpec.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpMessageHandlerSpec.java @@ -17,13 +17,16 @@ package org.springframework.integration.http.dsl; import java.net.URI; +import java.util.Arrays; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.http.client.ClientHttpRequestFactory; +import org.springframework.http.converter.HttpMessageConverter; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; import org.springframework.util.Assert; +import org.springframework.web.client.ResponseErrorHandler; import org.springframework.web.client.RestTemplate; /** @@ -66,8 +69,33 @@ public class HttpMessageHandlerSpec return this; } + /** + * Set the {@link ResponseErrorHandler} for the underlying {@link RestTemplate}. + * @param errorHandler The error handler. + * @return the spec + */ + public HttpMessageHandlerSpec errorHandler(ResponseErrorHandler errorHandler) { + Assert.isTrue(this.isClientSet(), + "the 'errorHandler' must be specified on the provided 'restTemplate'"); + this.target.setErrorHandler(errorHandler); + return _this(); + } + + /** + * Set a list of {@link HttpMessageConverter}s to be used by the underlying {@link RestTemplate}. + * Converters configured via this method will override the default converters. + * @param messageConverters The message converters. + * @return the spec + */ + public HttpMessageHandlerSpec messageConverters(HttpMessageConverter... messageConverters) { + Assert.isTrue(!isClientSet(), "the 'messageConverters' must be specified on the provided restTemplate"); + this.target.setMessageConverters(Arrays.asList(messageConverters)); + return _this(); + } + @Override - protected boolean isRestTemplateSet() { + protected boolean isClientSet() { return this.restTemplate != null; } + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java new file mode 100644 index 0000000000..898387106c --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpMessageHandlerSpec.java @@ -0,0 +1,60 @@ +/* + * 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.http.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.web.reactive.function.client.WebClient; + +/** + * The {@link BaseHttpMessageHandlerSpec} implementation for the {@link ReactiveHttpRequestExecutingMessageHandler}. + * + * @author Shiliang Li + * @author Artem Bilan + * + * @since 5.0 + * + * @see ReactiveHttpRequestExecutingMessageHandler + */ +public class ReactiveHttpMessageHandlerSpec + extends BaseHttpMessageHandlerSpec { + + private final WebClient webClient; + + ReactiveHttpMessageHandlerSpec(URI uri, WebClient webClient) { + this(new ValueExpression<>(uri), webClient); + } + + ReactiveHttpMessageHandlerSpec(String uri, WebClient webClient) { + this(new LiteralExpression(uri), webClient); + } + + ReactiveHttpMessageHandlerSpec(Expression uriExpression, WebClient webClient) { + super(new ReactiveHttpRequestExecutingMessageHandler(uriExpression, webClient)); + this.webClient = webClient; + } + + @Override + protected boolean isClientSet() { + return this.webClient != null; + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java index 7c82272001..2b026aa04f 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java @@ -36,7 +36,6 @@ import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; -import org.springframework.http.converter.HttpMessageConverter; import org.springframework.integration.expression.ExpressionEvalMap; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; @@ -44,6 +43,7 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand import org.springframework.integration.http.support.DefaultHttpHeaderMapper; import org.springframework.integration.mapping.HeaderMapper; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; @@ -53,7 +53,6 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.util.StringUtils; -import org.springframework.web.client.ResponseErrorHandler; import org.springframework.web.client.RestTemplate; import org.springframework.web.util.UriComponents; import org.springframework.web.util.UriComponentsBuilder; @@ -163,14 +162,13 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac * @return whether a reply Message is expected. * @see AbstractHttpRequestExecutingMessageHandler#setExpectReply(boolean) */ - public boolean getExpectReply() { + public boolean isExpectReply() { return this.expectReply; } /** * Specify whether a reply Message is expected. If not, this handler will simply return null for a * successful response or throw an Exception for a non-successful response. The default is true. - * * @param expectReply true if a reply is expected. */ public void setExpectReply(boolean expectReply) { @@ -204,21 +202,6 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac this.expectedResponseTypeExpression = expectedResponseTypeExpression; } - /** - * Set the {@link ResponseErrorHandler} for the underlying implementation. - * - * @param errorHandler The error handler. - */ - public abstract void setErrorHandler(ResponseErrorHandler errorHandler); - - /** - * Set a list of {@link HttpMessageConverter}s to be used by the underlying implementation. - * Converters configured via this method will override the default converters. - * - * @param messageConverters The message converters. - */ - public abstract void setMessageConverters(List> messageConverters); - /** * Set the {@link HeaderMapper} to use when mapping between HTTP headers and MessageHeaders. * @param headerMapper The header mapper. @@ -228,8 +211,6 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac this.headerMapper = headerMapper; } - - /** * Set the Map of URI variable expressions to evaluate against the outbound message * when replacing the variable placeholders in a URI template. @@ -280,16 +261,17 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac try { HttpMethod httpMethod = this.determineHttpMethod(requestMessage); - if (!this.shouldIncludeRequestBody(httpMethod) && this.extractPayloadExplicitlySet) { + if (!shouldIncludeRequestBody(httpMethod) && this.extractPayloadExplicitlySet) { if (logger.isWarnEnabled()) { - logger.warn("The 'extractPayload' attribute has no relevance for the current request since the HTTP Method is '" + - httpMethod + "', and no request body will be sent for that method."); + logger.warn("The 'extractPayload' attribute has no relevance for the current request " + + "since the HTTP Method is '" + httpMethod + + "', and no request body will be sent for that method."); } } - Object expectedResponseType = this.determineExpectedResponseType(requestMessage); + Object expectedResponseType = determineExpectedResponseType(requestMessage); - HttpEntity httpRequest = this.generateHttpRequest(requestMessage, httpMethod); + HttpEntity httpRequest = generateHttpRequest(requestMessage, httpMethod); Map uriVariables = this.determineUriVariables(requestMessage); UriComponentsBuilder uriComponentsBuilder = uri instanceof String ? UriComponentsBuilder.fromUriString((String) uri) @@ -297,7 +279,7 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac UriComponents uriComponents = uriComponentsBuilder.buildAndExpand(uriVariables); realUri = this.encodeUri ? uriComponents.toUri() : new URI(uriComponents.toUriString()); - return this.exchange(realUri, httpMethod, httpRequest, expectedResponseType); + return exchange(realUri, httpMethod, httpRequest, expectedResponseType, requestMessage); } catch (MessagingException e) { throw e; @@ -308,7 +290,8 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac } } - abstract protected Object exchange(URI realUri, HttpMethod httpMethod, HttpEntity httpRequest, Object expectedResponseType); + protected abstract Object exchange(URI realUri, HttpMethod httpMethod, HttpEntity httpRequest, + Object expectedResponseType, Message requestMessage); protected Object getReply(ResponseEntity httpResponse) { if (this.expectReply) { @@ -317,17 +300,21 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac if (this.transferCookies) { this.doConvertSetCookie(headers); } + AbstractIntegrationMessageBuilder replyBuilder = null; + MessageBuilderFactory messageBuilderFactory = getMessageBuilderFactory(); if (httpResponse.hasBody()) { Object responseBody = httpResponse.getBody(); - replyBuilder = (responseBody instanceof Message) ? - this.getMessageBuilderFactory().fromMessage((Message) responseBody) : this.getMessageBuilderFactory().withPayload(responseBody); + replyBuilder = (responseBody instanceof Message) + ? messageBuilderFactory.fromMessage((Message) responseBody) + : messageBuilderFactory.withPayload(responseBody); } else { - replyBuilder = this.getMessageBuilderFactory().withPayload(httpResponse); + replyBuilder = messageBuilderFactory.withPayload(httpResponse); } - replyBuilder.setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, httpResponse.getStatusCode()); + replyBuilder.setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, + httpResponse.getStatusCode()); return replyBuilder.copyHeaders(headers); } return null; diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandler.java deleted file mode 100644 index 2579265e0f..0000000000 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandler.java +++ /dev/null @@ -1,162 +0,0 @@ -/* - * 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.http.outbound; - -import java.net.URI; -import java.util.List; - -import org.springframework.beans.factory.BeanFactory; -import org.springframework.core.ParameterizedTypeReference; -import org.springframework.expression.Expression; -import org.springframework.expression.common.LiteralExpression; -import org.springframework.http.HttpEntity; -import org.springframework.http.HttpMethod; -import org.springframework.http.ResponseEntity; -import org.springframework.http.client.AsyncClientHttpRequestFactory; -import org.springframework.http.converter.HttpMessageConverter; -import org.springframework.integration.expression.ValueExpression; -import org.springframework.messaging.MessageHandler; -import org.springframework.util.Assert; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; -import org.springframework.web.client.AsyncRestTemplate; -import org.springframework.web.client.ResponseErrorHandler; - -/** - * A {@link MessageHandler} implementation that executes HTTP requests by delegating - * to an {@link AsyncRestTemplate} instance. - * @see HttpRequestExecutingMessageHandler - * @author Shiliang Li - * @since 5.0 - */ -public class AsyncHttpRequestExecutingMessageHandler extends AbstractHttpRequestExecutingMessageHandler { - - private final AsyncRestTemplate asyncRestTemplate; - - /** - * Create a handler that will send requests to the provided URI. - * - * @param uri The URI. - */ - public AsyncHttpRequestExecutingMessageHandler(URI uri) { - this(new ValueExpression<>(uri)); - } - - /** - * Create a handler that will send requests to the provided URI. - * - * @param uri The URI. - */ - public AsyncHttpRequestExecutingMessageHandler(String uri) { - this(uri, null); - } - - /** - * Create a handler that will send requests to the provided URI Expression. - * - * @param uriExpression The URI expression. - */ - public AsyncHttpRequestExecutingMessageHandler(Expression uriExpression) { - this(uriExpression, null); - } - - /** - * Create a handler that will send requests to the provided URI using a provided AsyncRestTemplate - * @param uri The URI. - * @param asyncRestTemplate The rest template. - */ - public AsyncHttpRequestExecutingMessageHandler(String uri, AsyncRestTemplate asyncRestTemplate) { - this(new LiteralExpression(uri), asyncRestTemplate); - /* - * We'd prefer to do this assertion first, but the compiler doesn't allow it. However, - * it's safe because the literal expression simply wraps the String variable, even - * when null. - */ - Assert.hasText(uri, "URI is required"); - } - - /** - * Create a handler that will send requests to the provided URI using a provided AsyncRestTemplate - * @param uriExpression A SpEL Expression that can be resolved against the message object and - * {@link BeanFactory}. - * @param asyncRestTemplate The rest template. - */ - public AsyncHttpRequestExecutingMessageHandler(Expression uriExpression, AsyncRestTemplate asyncRestTemplate) { - super(uriExpression); - this.asyncRestTemplate = (asyncRestTemplate == null ? new AsyncRestTemplate() : asyncRestTemplate); - this.setAsync(true); - } - - @Override - public String getComponentType() { - return (this.getExpectReply() ? "http:outbound-async-gateway" : "http:outbound-async-channel-adapter"); - } - - /** - * Set the {@link ResponseErrorHandler} for the underlying {@link AsyncRestTemplate}. - * - * @param errorHandler The error handler. - * - * @see AsyncRestTemplate#setErrorHandler(ResponseErrorHandler) - */ - @Override - public void setErrorHandler(ResponseErrorHandler errorHandler) { - this.asyncRestTemplate.setErrorHandler(errorHandler); - } - - /** - * Set a list of {@link HttpMessageConverter}s to be used by the underlying {@link AsyncRestTemplate}. - * Converters configured via this method will override the default converters. - * - * @param messageConverters The message converters. - * - * @see AsyncRestTemplate#setMessageConverters(java.util.List) - */ - @Override - public void setMessageConverters(List> messageConverters) { - this.asyncRestTemplate.setMessageConverters(messageConverters); - } - - /** - * Set the {@link AsyncClientHttpRequestFactory} for the underlying {@link AsyncRestTemplate}. - * - * @param asyncRequestFactory The request factory. - * - * @see AsyncRestTemplate#setAsyncRequestFactory(AsyncClientHttpRequestFactory) - */ - public void setAsyncRequestFactory(AsyncClientHttpRequestFactory asyncRequestFactory) { - this.asyncRestTemplate.setAsyncRequestFactory(asyncRequestFactory); - } - - @Override - protected Object exchange(URI uri, HttpMethod httpMethod, HttpEntity httpRequest, Object expectedResponseType) { - SettableListenableFuture replyMessageFuture = new SettableListenableFuture<>(); - ListenableFuture> responseFuture; - if (expectedResponseType instanceof ParameterizedTypeReference) { - responseFuture = this.asyncRestTemplate.exchange(uri, httpMethod, httpRequest, (ParameterizedTypeReference) expectedResponseType); - } - else { - responseFuture = this.asyncRestTemplate.exchange(uri, httpMethod, httpRequest, (Class) expectedResponseType); - } - - responseFuture.addCallback( - result -> replyMessageFuture.set(getReply(result)), - replyMessageFuture::setException); - - return replyMessageFuture; - } -} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/HttpRequestExecutingMessageHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/HttpRequestExecutingMessageHandler.java index 8286030ca9..8846a78afc 100755 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/HttpRequestExecutingMessageHandler.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/HttpRequestExecutingMessageHandler.java @@ -31,6 +31,7 @@ import org.springframework.http.client.ClientHttpRequestFactory; import org.springframework.http.converter.HttpMessageConverter; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; import org.springframework.web.client.ResponseErrorHandler; @@ -53,6 +54,7 @@ import org.springframework.web.client.RestTemplate; * @author Artem Bilan * @author Wallace Wadge * @author Shiliang Li + * * @since 2.0 */ public class HttpRequestExecutingMessageHandler extends AbstractHttpRequestExecutingMessageHandler { @@ -113,17 +115,14 @@ public class HttpRequestExecutingMessageHandler extends AbstractHttpRequestExecu @Override public String getComponentType() { - return (this.getExpectReply() ? "http:outbound-gateway" : "http:outbound-channel-adapter"); + return (this.isExpectReply() ? "http:outbound-gateway" : "http:outbound-channel-adapter"); } /** * Set the {@link ResponseErrorHandler} for the underlying {@link RestTemplate}. - * * @param errorHandler The error handler. - * * @see RestTemplate#setErrorHandler(ResponseErrorHandler) */ - @Override public void setErrorHandler(ResponseErrorHandler errorHandler) { this.restTemplate.setErrorHandler(errorHandler); } @@ -131,12 +130,9 @@ public class HttpRequestExecutingMessageHandler extends AbstractHttpRequestExecu /** * Set a list of {@link HttpMessageConverter}s to be used by the underlying {@link RestTemplate}. * Converters configured via this method will override the default converters. - * * @param messageConverters The message converters. - * * @see RestTemplate#setMessageConverters(java.util.List) */ - @Override public void setMessageConverters(List> messageConverters) { this.restTemplate.setMessageConverters(messageConverters); } @@ -153,10 +149,12 @@ public class HttpRequestExecutingMessageHandler extends AbstractHttpRequestExecu } @Override - protected Object exchange(URI uri, HttpMethod httpMethod, HttpEntity httpRequest, Object expectedResponseType) { + protected Object exchange(URI uri, HttpMethod httpMethod, HttpEntity httpRequest, Object expectedResponseType, + Message requestMessage) { ResponseEntity httpResponse; if (expectedResponseType instanceof ParameterizedTypeReference) { - httpResponse = this.restTemplate.exchange(uri, httpMethod, httpRequest, (ParameterizedTypeReference) expectedResponseType); + httpResponse = this.restTemplate.exchange(uri, httpMethod, httpRequest, + (ParameterizedTypeReference) expectedResponseType); } else { httpResponse = this.restTemplate.exchange(uri, httpMethod, httpRequest, (Class) expectedResponseType); diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java new file mode 100644 index 0000000000..ad07f7fe08 --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandler.java @@ -0,0 +1,198 @@ +/* + * 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.http.outbound; + +import java.net.URI; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ResolvableType; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; +import org.springframework.web.reactive.function.BodyExtractors; +import org.springframework.web.reactive.function.BodyInserters; +import org.springframework.web.reactive.function.client.ClientResponse; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.reactive.function.client.WebClientException; + +import reactor.core.publisher.Mono; + +/** + * A {@link MessageHandler} implementation that executes HTTP requests by delegating + * to a Reactive {@link WebClient} instance. + * + * @author Shiliang Li + * @author Artem Bilan + * + * @since 5.0 + * + * @see HttpRequestExecutingMessageHandler + */ +public class ReactiveHttpRequestExecutingMessageHandler extends AbstractHttpRequestExecutingMessageHandler { + + private final WebClient webClient; + + /** + * Create a handler that will send requests to the provided URI. + * @param uri The URI. + */ + public ReactiveHttpRequestExecutingMessageHandler(URI uri) { + this(new ValueExpression<>(uri)); + } + + /** + * Create a handler that will send requests to the provided URI. + * @param uri The URI. + */ + public ReactiveHttpRequestExecutingMessageHandler(String uri) { + this(uri, null); + } + + /** + * Create a handler that will send requests to the provided URI Expression. + * @param uriExpression The URI expression. + */ + public ReactiveHttpRequestExecutingMessageHandler(Expression uriExpression) { + this(uriExpression, null); + } + + /** + * Create a handler that will send requests to the provided URI using a provided WebClient. + * @param uri The URI. + * @param webClient The WebClient to use. + */ + public ReactiveHttpRequestExecutingMessageHandler(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, + * it's safe because the literal expression simply wraps the String variable, even + * when null. + */ + Assert.hasText(uri, "URI is required"); + } + + /** + * Create a handler that will send requests to the provided URI using a provided WebClient. + * @param uriExpression A SpEL Expression that can be resolved against the message object and + * {@link BeanFactory}. + * @param webClient The WebClient to use. + */ + public ReactiveHttpRequestExecutingMessageHandler(Expression uriExpression, WebClient webClient) { + super(uriExpression); + this.webClient = (webClient == null ? WebClient.create() : webClient); + this.setAsync(true); + } + + @Override + public String getComponentType() { + return (isExpectReply() ? "http:outbound-reactive-gateway" : "http:outbound-reactive-channel-adapter"); + } + + @Override + protected Object exchange(URI uri, HttpMethod httpMethod, HttpEntity httpRequest, Object expectedResponseType, + Message requestMessage) { + WebClient.UriSpec uriSpec; + + // TODO use WebClient.method(HttpMethod) in the future version + + switch (httpMethod) { + case GET: + uriSpec = this.webClient.get(); + break; + case HEAD: + uriSpec = this.webClient.head(); + break; + case POST: + uriSpec = this.webClient.post(); + break; + case PUT: + uriSpec = this.webClient.put(); + break; + case PATCH: + uriSpec = this.webClient.patch(); + break; + case DELETE: + uriSpec = this.webClient.delete(); + break; + case OPTIONS: + uriSpec = this.webClient.options(); + break; + case TRACE: + throw new UnsupportedOperationException("WebClient doesn't support the TRACE HTTP method"); + default: + throw new UnsupportedOperationException("Unsupported HTTP method"); + } + + WebClient.HeaderSpec spec = uriSpec.uri(uri) + .headers(httpRequest.getHeaders()); + + Mono responseMono; + if (httpRequest.hasBody()) { + responseMono = spec.exchange(BodyInserters.fromObject(httpRequest.getBody())); + } + else { + responseMono = spec.exchange(); + } + + if (isExpectReply()) { + + ResolvableType responseType; + + if (expectedResponseType instanceof ParameterizedTypeReference) { + responseType = ResolvableType.forType(((ParameterizedTypeReference) expectedResponseType).getType()); + } + else if (expectedResponseType != null) { + responseType = ResolvableType.forClass((Class) expectedResponseType); + } + else { + responseType = null; + } + + return responseMono + .map(response -> + new ResponseEntity<>(responseType != null + ? response.body(BodyExtractors.toMono(responseType)).block() + : null, + response.headers().asHttpHeaders(), + response.statusCode())) + .map(this::getReply); + } + else { + responseMono + .doOnNext(response -> { + HttpStatus httpStatus = response.statusCode(); + if (httpStatus.is4xxClientError() || httpStatus.is5xxServerError()) { + throw new WebClientException( + "ClientResponse has erroneous status code: " + httpStatus.value() + + " " + httpStatus.getReasonPhrase()); + } + }) + .subscribe(v -> { }, ex -> sendErrorMessage(requestMessage, ex)); + + return null; + } + } + +} 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 486d963554..a7c006b2a1 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,11 +401,11 @@ - + Configures a Consumer Endpoint for the - 'org.springframework.integration.http.outbound.AsyncHttpRequestExecutingMessageHandler' + 'org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler' with 'expectReply = false' that sends HTTP requests based on incoming messages. @@ -424,7 +424,19 @@ - + + + + + + + + + A reference to an org.springframework.web.reactive.function.client.WebClient bean + which is used to send to send the HTTP Requests reactive manner. + + + @@ -532,11 +544,11 @@ - + Configures a Consumer Endpoint for the - 'org.springframework.integration.http.outbound.AsyncHttpRequestExecutingMessageHandler' + 'org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler' that sends HTTP requests based on incoming messages and expects HTTP responses. @@ -551,8 +563,10 @@ - - + + @@ -620,7 +634,19 @@ - + + + + + + + + + A reference to an org.springframework.web.reactive.function.client.WebClient bean + which is used to send to send the HTTP Requests reactive manner. + + + @@ -859,33 +885,6 @@ - - - - - - - - - - The reference to org.springframework.web.client.AsyncRestTemplate bean to send the HTTP Request. - - - - - - - - - - - - - - - @@ -911,6 +910,27 @@ + + + + Reference to a ResponseErrorHandler to be used by the underlying RestTemplate. + + + + + + + + + + + + Provide a reference to a list of HttpMessageConverter instances. If specified, + these converters will replace all of the default converters that would normally + be present on the underlying RestTemplate. + + + @@ -992,14 +1012,6 @@ - - - - Provide a reference to a list of HttpMessageConverter instances. If specified, these converters will replace - all of the default converters that would normally be present on the underlying RestTemplate. - - - @@ -1024,18 +1036,6 @@ ]]> - - - - - - - - - - @@ -14,29 +14,32 @@ - + - + - + - + + url="http://localhost/test2/{foo}" + http-method="GET" + channel="requests" + charset="UTF-8" + message-converters="converterList" + extract-payload="false" + expected-response-type="java.lang.Boolean" + mapped-request-headers="requestHeader1, requestHeader2" + request-factory="testRequestFactory" + error-handler="testErrorHandler" + order="77" + auto-startup="false"> @@ -47,21 +50,22 @@ + url="http://localhost/test1" channel="requests" + uri-variables-expression="@uriVariables" + rest-template="customRestTemplate"/> - + + url-expression="'http://localhost/test1'" channel="requests" + rest-template="customRestTemplate"/> @@ -102,7 +106,8 @@ - + diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests.java index 5233f420c0..d83400b55c 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests.java @@ -41,14 +41,12 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.http.HttpMethod; -import org.springframework.http.client.AsyncClientHttpRequestFactory; import org.springframework.http.client.ClientHttpRequestFactory; import org.springframework.http.client.ClientHttpResponse; import org.springframework.http.client.SimpleClientHttpRequestFactory; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; -import org.springframework.integration.http.outbound.AsyncHttpRequestExecutingMessageHandler; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -58,9 +56,9 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ObjectUtils; -import org.springframework.web.client.AsyncRestTemplate; import org.springframework.web.client.ResponseErrorHandler; import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; /** * @author Mark Fisher @@ -75,40 +73,51 @@ import org.springframework.web.client.RestTemplate; @DirtiesContext public class HttpOutboundChannelAdapterParserTests { - @Autowired @Qualifier("minimalConfig") + @Autowired + @Qualifier("minimalConfig") private AbstractEndpoint minimalConfig; - @Autowired @Qualifier("fullConfig") + @Autowired + @Qualifier("fullConfig") private AbstractEndpoint fullConfig; - @Autowired @Qualifier("restTemplateConfig") + @Autowired + @Qualifier("restTemplateConfig") private AbstractEndpoint restTemplateConfig; - @Autowired @Qualifier("asyncMinimalConfig") - private AbstractEndpoint asyncMinimalConfig; + @Autowired + @Qualifier("reactiveMinimalConfig") + private AbstractEndpoint reactiveMinimalConfig; - @Autowired @Qualifier("asyncRestTemplateConfig") - private AbstractEndpoint asyncRestTemplateConfig; + @Autowired + @Qualifier("reactiveWebClientConfig") + private AbstractEndpoint reactiveWebClientConfig; - @Autowired @Qualifier("customRestTemplate") + @Autowired + @Qualifier("customRestTemplate") private RestTemplate customRestTemplate; - @Autowired @Qualifier("asyncRestTemplate") - private AsyncRestTemplate asyncRestTemplate; + @Autowired + private WebClient webClient; - @Autowired @Qualifier("withUrlAndTemplate") + @Autowired + @Qualifier("withUrlAndTemplate") private AbstractEndpoint withUrlAndTemplate; - @Autowired @Qualifier("withUrlExpression") + @Autowired + @Qualifier("withUrlExpression") private AbstractEndpoint withUrlExpression; - @Autowired @Qualifier("withAdvice") + @Autowired + @Qualifier("withAdvice") private AbstractEndpoint withAdvice; - @Autowired @Qualifier("withUrlExpressionAndTemplate") + @Autowired + @Qualifier("withUrlExpressionAndTemplate") private AbstractEndpoint withUrlExpressionAndTemplate; - @Autowired @Qualifier("withPoller1") + @Autowired + @Qualifier("withPoller1") private AbstractEndpoint withPoller1; @Autowired @@ -120,7 +129,7 @@ public class HttpOutboundChannelAdapterParserTests { public void minimalConfig() { DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.minimalConfig); RestTemplate restTemplate = - TestUtils.getPropertyValue(this.minimalConfig, "handler.restTemplate", RestTemplate.class); + TestUtils.getPropertyValue(this.minimalConfig, "handler.restTemplate", RestTemplate.class); assertNotSame(customRestTemplate, restTemplate); HttpRequestExecutingMessageHandler handler = (HttpRequestExecutingMessageHandler) endpointAccessor.getPropertyValue("handler"); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); @@ -179,23 +188,20 @@ public class HttpOutboundChannelAdapterParserTests { } @Test - public void asyncMinimalConfig() { - DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.asyncMinimalConfig); - AsyncRestTemplate asyncRestTemplate = - TestUtils.getPropertyValue(this.asyncMinimalConfig, "handler.asyncRestTemplate", AsyncRestTemplate.class); - assertNotSame(this.asyncRestTemplate, asyncRestTemplate); - AsyncHttpRequestExecutingMessageHandler handler = (AsyncHttpRequestExecutingMessageHandler) endpointAccessor.getPropertyValue("handler"); + 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")); - DirectFieldAccessor templateAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("asyncRestTemplate")); - AsyncClientHttpRequestFactory asyncRequestFactory = (AsyncClientHttpRequestFactory) - templateAccessor.getPropertyValue("asyncRequestFactory"); - assertTrue(asyncRequestFactory instanceof SimpleClientHttpRequestFactory); Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression"); assertEquals("http://localhost/test1", uriExpression.getValue()); - assertEquals(HttpMethod.POST.name(), TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); + assertEquals(HttpMethod.POST.name(), + TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); assertEquals(true, handlerAccessor.getPropertyValue("extractPayload")); } @@ -203,16 +209,13 @@ public class HttpOutboundChannelAdapterParserTests { @Test public void restTemplateConfig() { RestTemplate restTemplate = - TestUtils.getPropertyValue(this.restTemplateConfig, "handler.restTemplate", RestTemplate.class); + TestUtils.getPropertyValue(this.restTemplateConfig, "handler.restTemplate", RestTemplate.class); assertEquals(customRestTemplate, restTemplate); } @Test - public void asyncRestTemplateConfig() { - AsyncRestTemplate asyncRestTemplate = TestUtils.getPropertyValue( - this.asyncRestTemplateConfig, - "handler.asyncRestTemplate", AsyncRestTemplate.class); - assertSame(this.asyncRestTemplate, asyncRestTemplate); + public void reactiveWebClientConfig() { + assertSame(this.webClient, TestUtils.getPropertyValue(this.reactiveWebClientConfig, "handler.webClient")); } @Test(expected = BeanDefinitionParsingException.class) @@ -225,7 +228,7 @@ public class HttpOutboundChannelAdapterParserTests { public void withUrlAndTemplate() { DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.withUrlAndTemplate); RestTemplate restTemplate = - TestUtils.getPropertyValue(this.withUrlAndTemplate, "handler.restTemplate", RestTemplate.class); + TestUtils.getPropertyValue(this.withUrlAndTemplate, "handler.restTemplate", RestTemplate.class); assertSame(customRestTemplate, restTemplate); HttpRequestExecutingMessageHandler handler = (HttpRequestExecutingMessageHandler) endpointAccessor.getPropertyValue("handler"); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); @@ -255,7 +258,7 @@ public class HttpOutboundChannelAdapterParserTests { public void withUrlExpression() { DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.withUrlExpression); RestTemplate restTemplate = - TestUtils.getPropertyValue(this.withUrlExpression, "handler.restTemplate", RestTemplate.class); + TestUtils.getPropertyValue(this.withUrlExpression, "handler.restTemplate", RestTemplate.class); assertNotSame(customRestTemplate, restTemplate); HttpRequestExecutingMessageHandler handler = (HttpRequestExecutingMessageHandler) endpointAccessor.getPropertyValue("handler"); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); @@ -285,7 +288,7 @@ public class HttpOutboundChannelAdapterParserTests { public void withUrlExpressionAndTemplate() { DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.withUrlExpressionAndTemplate); RestTemplate restTemplate = - TestUtils.getPropertyValue(this.withUrlExpressionAndTemplate, "handler.restTemplate", RestTemplate.class); + TestUtils.getPropertyValue(this.withUrlExpressionAndTemplate, "handler.restTemplate", RestTemplate.class); assertSame(customRestTemplate, restTemplate); HttpRequestExecutingMessageHandler handler = (HttpRequestExecutingMessageHandler) endpointAccessor.getPropertyValue("handler"); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); @@ -325,6 +328,7 @@ public class HttpOutboundChannelAdapterParserTests { @Override public void handleError(ClientHttpResponse response) throws IOException { } + } public static class FooAdvice extends AbstractRequestHandlerAdvice { @@ -336,4 +340,5 @@ public class HttpOutboundChannelAdapterParserTests { } } + } diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests-context.xml b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests-context.xml index d4177c3294..74c247e299 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests-context.xml +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests-context.xml @@ -1,10 +1,10 @@ - + @@ -23,46 +24,44 @@ + url="http://localhost/test2" + http-method="PUT" + request-channel="requests" + request-factory="testRequestFactory" + reply-timeout="1234" + message-converters="converterList" + extract-request-payload="false" + expected-response-type="java.lang.String" + mapped-request-headers="requestHeader1, requestHeader2" + mapped-response-headers="responseHeader" + error-handler="testErrorHandler" + reply-channel="replies" + charset="UTF-8" + order="77" + auto-startup="false" + transfer-cookies="true"> - + - + - + @@ -74,7 +73,7 @@ - + @@ -116,7 +115,8 @@ - + 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 8548859507..4a1ce768ce 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * 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. @@ -19,6 +19,7 @@ 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; @@ -41,14 +42,12 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.http.HttpMethod; -import org.springframework.http.client.AsyncClientHttpRequestFactory; import org.springframework.http.client.ClientHttpRequestFactory; import org.springframework.http.client.ClientHttpResponse; import org.springframework.http.client.SimpleClientHttpRequestFactory; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; -import org.springframework.integration.http.outbound.AsyncHttpRequestExecutingMessageHandler; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -58,8 +57,8 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ObjectUtils; -import org.springframework.web.client.AsyncRestTemplate; import org.springframework.web.client.ResponseErrorHandler; +import org.springframework.web.reactive.function.client.WebClient; /** * @author Mark Fisher @@ -72,29 +71,36 @@ import org.springframework.web.client.ResponseErrorHandler; @DirtiesContext public class HttpOutboundGatewayParserTests { - @Autowired @Qualifier("minimalConfig") + @Autowired + @Qualifier("minimalConfig") private AbstractEndpoint minimalConfigEndpoint; - @Autowired @Qualifier("fullConfig") + @Autowired + @Qualifier("fullConfig") private AbstractEndpoint fullConfigEndpoint; - @Autowired @Qualifier("asyncMinimalConfig") - private AbstractEndpoint asyncMinimalConfigEndpoint; + @Autowired + @Qualifier("reactiveMinimalConfig") + private AbstractEndpoint reactiveMinimalConfigEndpoint; - @Autowired @Qualifier("asyncFullConfig") - private AbstractEndpoint asyncFullConfigEndpoint; + @Autowired + @Qualifier("reactiveFullConfig") + private AbstractEndpoint reactiveFullConfigEndpoint; - @Autowired @Qualifier("withUrlExpression") + @Autowired + @Qualifier("withUrlExpression") private AbstractEndpoint withUrlExpressionEndpoint; - @Autowired @Qualifier("withAdvice") + @Autowired + @Qualifier("withAdvice") private AbstractEndpoint withAdvice; - @Autowired @Qualifier("withPoller1") + @Autowired + @Qualifier("withPoller1") private AbstractEndpoint withPoller1; - @Autowired @Qualifier("asyncRestTemplate") - private AsyncRestTemplate asyncRestTemplate; + @Autowired + private WebClient webClient; @Autowired private ApplicationContext applicationContext; @@ -173,22 +179,19 @@ public class HttpOutboundGatewayParserTests { } @Test - public void asyncMinimalConfig() { - AsyncHttpRequestExecutingMessageHandler handler = (AsyncHttpRequestExecutingMessageHandler) new DirectFieldAccessor( - this.asyncMinimalConfigEndpoint).getPropertyValue("handler"); - MessageChannel requestChannel = (MessageChannel) new DirectFieldAccessor( - this.minimalConfigEndpoint).getPropertyValue("inputChannel"); + 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); - DirectFieldAccessor templateAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("asyncRestTemplate")); - AsyncClientHttpRequestFactory requestFactory = (AsyncClientHttpRequestFactory) - templateAccessor.getPropertyValue("asyncRequestFactory"); - assertTrue(requestFactory instanceof SimpleClientHttpRequestFactory); + 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(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")); @@ -196,11 +199,11 @@ public class HttpOutboundGatewayParserTests { @Test @SuppressWarnings("unchecked") - public void asyncFullConfig() { - DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.asyncFullConfigEndpoint); - AsyncHttpRequestExecutingMessageHandler handler = (AsyncHttpRequestExecutingMessageHandler) endpointAccessor.getPropertyValue("handler"); + public void reactiveFullConfig() { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfigEndpoint); + Object handler = endpointAccessor.getPropertyValue("handler"); MessageChannel requestChannel = (MessageChannel) new DirectFieldAccessor( - this.asyncFullConfigEndpoint).getPropertyValue("inputChannel"); + this.reactiveFullConfigEndpoint).getPropertyValue("inputChannel"); assertEquals(this.applicationContext.getBean("requests"), requestChannel); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); assertEquals(77, handlerAccessor.getPropertyValue("order")); @@ -208,24 +211,15 @@ public class HttpOutboundGatewayParserTests { Object replyChannel = handlerAccessor.getPropertyValue("outputChannel"); assertNotNull(replyChannel); assertEquals(this.applicationContext.getBean("replies"), replyChannel); - DirectFieldAccessor asyncTemplateAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("asyncRestTemplate")); - DirectFieldAccessor syncTemplateAccessor = new DirectFieldAccessor(asyncTemplateAccessor.getPropertyValue("syncTemplate")); - AsyncClientHttpRequestFactory requestFactory = (AsyncClientHttpRequestFactory) - asyncTemplateAccessor.getPropertyValue("asyncRequestFactory"); - assertTrue(requestFactory instanceof SimpleClientHttpRequestFactory); - Object converterListBean = this.applicationContext.getBean("converterList"); - assertEquals(converterListBean, syncTemplateAccessor.getPropertyValue("messageConverters")); - assertEquals(String.class.getName(), TestUtils.getPropertyValue(handler, "expectedResponseTypeExpression", Expression.class).getValue()); + 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(HttpMethod.PUT.name(), + TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString()); assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset")); assertEquals(false, handlerAccessor.getPropertyValue("extractPayload")); - Object requestFactoryBean = this.applicationContext.getBean("testRequestFactory"); - assertEquals(requestFactoryBean, requestFactory); - Object errorHandlerBean = this.applicationContext.getBean("testErrorHandler"); - assertEquals(errorHandlerBean, syncTemplateAccessor.getPropertyValue("errorHandler")); Object sendTimeout = new DirectFieldAccessor( handlerAccessor.getPropertyValue("messagingTemplate")).getPropertyValue("sendTimeout"); assertEquals(new Long("1234"), sendTimeout); @@ -302,7 +296,6 @@ public class HttpOutboundGatewayParserTests { } - public static class StubErrorHandler implements ResponseErrorHandler { @Override @@ -324,4 +317,5 @@ public class HttpOutboundGatewayParserTests { } } + } 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 316a260208..84b2603363 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 @@ -18,16 +18,12 @@ package org.springframework.integration.http.dsl; 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.client.match.MockRestRequestMatchers.method; -import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; -import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess; 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 java.util.List; -import org.hamcrest.Matchers; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -37,13 +33,15 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; 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.AsyncHttpRequestExecutingMessageHandler; 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.MessageChannel; @@ -58,16 +56,18 @@ 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.client.MockRestServiceServer; +import org.springframework.test.web.reactive.server.HttpHandlerConnector; 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.AsyncRestTemplate; import org.springframework.web.client.RestTemplate; import org.springframework.web.context.WebApplicationContext; +import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.servlet.config.annotation.EnableWebMvc; import org.springframework.web.util.UriComponentsBuilder; +import reactor.core.publisher.Mono; + /** * @author Artem Bilan * @author Shiliang Li @@ -86,7 +86,7 @@ public class HttpDslTests { private HttpRequestExecutingMessageHandler serviceInternalGatewayHandler; @Autowired - private AsyncHttpRequestExecutingMessageHandler serviceInternalAsyncGatewayHandler; + private ReactiveHttpRequestExecutingMessageHandler serviceInternalReactiveGatewayHandler; private MockMvc mockMvc; @@ -115,16 +115,21 @@ public class HttpDslTests { } @Test - public void testHttpAsyncProxyFlow() throws Exception { - AsyncRestTemplate asyncRestTemplate = new AsyncRestTemplate(); - String destinationUri = "http://www.springsource.org/spring-integration"; - MockRestServiceServer - .createServer(asyncRestTemplate) - .expect(requestTo(Matchers.startsWith(destinationUri))) - .andExpect(method(HttpMethod.POST)) - .andRespond(withSuccess("FOO", MediaType.TEXT_PLAIN)); - new DirectFieldAccessor(this.serviceInternalAsyncGatewayHandler) - .setPropertyValue("asyncRestTemplate", asyncRestTemplate); + 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(response::setComplete); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + new DirectFieldAccessor(this.serviceInternalReactiveGatewayHandler) + .setPropertyValue("webClient", webClient); this.mockMvc.perform( get("/service2") @@ -191,17 +196,18 @@ public class HttpDslTests { } @Bean - public IntegrationFlow httpAsyncProxyFlow() { + public IntegrationFlow httpReactiveProxyFlow() { return IntegrationFlows .from(Http.inboundGateway("/service2") .requestMapping(r -> r.params("name"))) - .handle(Http.>outboundAsyncGateway(m -> + .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("serviceInternalAsyncGateway")) + e -> e.id("serviceInternalReactiveGateway")) .get(); } diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandlerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandlerTests.java deleted file mode 100644 index 272dc8b634..0000000000 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/AsyncHttpRequestExecutingMessageHandlerTests.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * 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.http.outbound; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.springframework.test.web.client.match.MockRestRequestMatchers.method; -import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; -import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess; - -import org.junit.Test; -import org.springframework.http.HttpMethod; -import org.springframework.http.HttpStatus; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dsl.channel.MessageChannels; -import org.springframework.integration.http.HttpHeaders; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.test.web.client.MockRestServiceServer; -import org.springframework.web.client.AsyncRestTemplate; - -/** - * @author Shiliang Li - * @since 5.0 - */ -public class AsyncHttpRequestExecutingMessageHandlerTests { - - @Test - public void testAsyncReturn() { - AsyncRestTemplate asyncRestTemplate = new AsyncRestTemplate(); - String destinationUri = "http://www.springsource.org/spring-integration"; - MockRestServiceServer - .createServer(asyncRestTemplate) - .expect(requestTo(destinationUri)) - .andExpect(method(HttpMethod.POST)) - .andRespond(withSuccess()); - - AsyncHttpRequestExecutingMessageHandler asyncHandler = new AsyncHttpRequestExecutingMessageHandler( - destinationUri, - asyncRestTemplate); - QueueChannel ackChannel = MessageChannels.queue().get(); - asyncHandler.setOutputChannel(ackChannel); - asyncHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); - Message ack = ackChannel.receive(1000); - assertNotNull(ack); - assertNotNull(ack.getHeaders()); - assertEquals(ack.getHeaders().get(HttpHeaders.STATUS_CODE), HttpStatus.OK); - } -} diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java new file mode 100644 index 0000000000..e1f2dc4c36 --- /dev/null +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java @@ -0,0 +1,136 @@ +/* + * 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.http.outbound; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import java.time.Duration; + +import org.junit.Test; + +import org.springframework.http.HttpStatus; +import org.springframework.http.client.reactive.ClientHttpConnector; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.ReactiveChannel; +import org.springframework.integration.http.HttpHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.test.web.reactive.server.HttpHandlerConnector; +import org.springframework.web.reactive.function.client.WebClient; + +import reactor.core.publisher.Mono; + +/** + * @author Shiliang Li + * @author Artem Bilan + * + * @since 5.0 + */ +public class ReactiveHttpRequestExecutingMessageHandlerTests { + + @Test + public void testReactiveReturn() throws Throwable { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + response.setStatusCode(HttpStatus.OK); + return Mono.empty() + .then(response::setComplete); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + String destinationUri = "http://www.springsource.org/spring-integration"; + ReactiveHttpRequestExecutingMessageHandler reactiveHandler = + new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); + + ReactiveChannel ackChannel = new ReactiveChannel(); + reactiveHandler.setOutputChannel(ackChannel); + reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); + + Message ack = Mono.from(ackChannel).block(Duration.ofSeconds(10)); + + assertNotNull(ack); + assertNotNull(ack.getHeaders()); + assertEquals(ack.getHeaders().get(HttpHeaders.STATUS_CODE), HttpStatus.OK); + } + + @Test + public void testReactiveErrorOneWay() throws Throwable { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + response.setStatusCode(HttpStatus.UNAUTHORIZED); + return Mono.empty() + .then(response::setComplete); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + String destinationUri = "http://www.springsource.org/spring-integration"; + ReactiveHttpRequestExecutingMessageHandler reactiveHandler = + new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); + reactiveHandler.setExpectReply(false); + + QueueChannel errorChannel = new QueueChannel(); + reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world") + .setErrorChannel(errorChannel) + .build()); + + Message errorMessage = errorChannel.receive(10000); + + assertNotNull(errorMessage); + assertThat(errorMessage, instanceOf(ErrorMessage.class)); + Throwable throwable = (Throwable) errorMessage.getPayload(); + assertThat(throwable.getMessage(), containsString("401 Unauthorized")); + } + + @Test + public void testReactiveConnectErrorOneWay() throws Throwable { + ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { + throw new RuntimeException("Intentional connection error"); + }); + + WebClient webClient = WebClient.builder() + .clientConnector(httpConnector) + .build(); + + String destinationUri = "http://www.springsource.org/spring-integration"; + ReactiveHttpRequestExecutingMessageHandler reactiveHandler = + new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); + reactiveHandler.setExpectReply(false); + + QueueChannel errorChannel = new QueueChannel(); + reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world") + .setErrorChannel(errorChannel) + .build()); + + Message errorMessage = errorChannel.receive(10000); + + assertNotNull(errorMessage); + assertThat(errorMessage, instanceOf(ErrorMessage.class)); + Throwable throwable = (Throwable) errorMessage.getPayload(); + assertThat(throwable.getMessage(), containsString("Intentional connection error")); + } + + +} diff --git a/src/reference/asciidoc/http.adoc b/src/reference/asciidoc/http.adoc index cb6d4981d7..53d70662d4 100644 --- a/src/reference/asciidoc/http.adoc +++ b/src/reference/asciidoc/http.adoc @@ -5,7 +5,7 @@ === 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 `AsyncHttpRequestExecutingMessageHandler` +the HTTP support consists of the following gateway implementations: `HttpInboundEndpoint`, `HttpRequestExecutingMessageHandler` and `ReactiveHttpRequestExecutingMessageHandler` [[http-inbound]] === Http Inbound Components @@ -191,36 +191,42 @@ 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`). ===== -==== AsyncHttpRequestExecutingMessageHandler +==== ReactiveHttpRequestExecutingMessageHandler -The `AsyncHttpRequestExecutingMessageHandler` implementation is very similar to `HttpRequestExecutingMessageHandler` instead of delegating to a `AsyncRestTemplate`. +The `ReactiveHttpRequestExecutingMessageHandler` implementation is very similar to `HttpRequestExecutingMessageHandler` instead of delegating to a `WebClient` from Spring Framework WebFlux module. To configure it, write a bean like this: [source,xml] ---- - + ---- -You can configure the `AsyncClientHttpRequestFactory` instance to use: +You can configure a `WebClient` instance to use: [source,xml] ---- - + + + + - ---- -By default the HTTP request will be generated using an instance of `SimpleClientHttpRequestFactory`. -Use of the Apache Commons HTTP Async Client is also supported through the provided `HttpComponentsAsyncClientHttpRequestFactory` which can be injected as shown above. +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`. -For other settings like cookie, converters, etc, please see <> above. +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 <> above. [[http-namespace]] === HTTP Namespace Support @@ -544,29 +550,27 @@ The configuration looks very similar to the gateway: auto-startup="false"/> ---- -If you want to execute the http request in an asynchronous way, you can use the `outbound-async-gateway` or `outbound-async-channel-adapter`. +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] ---- - - diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 1f9faa2107..9f3c9cdab9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -27,10 +27,10 @@ See <> for more information. The new `MongoDbOutboundGateway` allows you to make queries to the database on demand by sending a message to its request channel. See <> for more information. -==== HTTP Async Outbound Gateway and Channel Adapter +==== HTTP Reactive Outbound Gateway and Channel Adapter -The new `AsyncHttpRequestExecutingMessageHandler` adds support for `AsyncRestTemplate` for outbound channel adapter and gateway. -See <> for more information. +The new `ReactiveHttpRequestExecutingMessageHandler` adds support for WebFlux `WebClient` for outbound channel adapter and gateway. +See <> for more information. ==== Content Type Conversion