From d4a99919ed78c011dc8d69bae745eedd425eb672 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 20 Jun 2017 10:14:57 -0400 Subject: [PATCH] INT-4300: Add WebFlux Server Support JIRA: https://jira.spring.io/browse/INT-4300 * Add `ReactiveHttpInboundEndpoint` based on the WebFlux foundation * Extract `BaseHttpInboundEndpoint` for common HTTP Inbound Channel Adapters options * Make `spring-webmvc` and `spring-webflux` as `optional` dependencies to let end-user to choose * Refactor `HttpContextUtils` to include constants for newly added WebFlux support * Introduce `BaseHttpInboundEndpoint.setRequestPayloadTypeClass()` for raw `Class` and modify existing `setRequestPayloadType()` for the `ResolvableType` * Refactor existing MVC tests and XML components parsers to use new `setRequestPayloadTypeClass()` * Add `MessagingGatewaySupport.sendAndReceiveMessageReactive()` to get a reply from downstream flow reactive back-pressure manner * Add `IntegrationHandlerResultHandler` implementation to let WebFlux infrastructure to handle the `Mono` from the `ReactiveHttpInboundEndpoint` properly * Fix `JdbcLockRegistryLeaderInitiatorTests` race condition to assert the `initiator1` is elected eventually after yielding when the `initiator2` is stopped * Fix JavaDocs issue in the `HttpRequestHandlingMessagingGateway` * Move all the "hard" logic in the `MessagingGatewaySupport#doSendAndReceiveMessageReactive` to the `Mono` chain ensuring back-pressure when `sendAndReceiveMessageReactive()` is called not from the Reactive Stream Add test-case to demonstrate SSE JIRA: https://jira.spring.io/browse/INT-3625 Some polishing and optimization for the `MessagingGatewaySupport.doSendAndReceiveMessageReactive()` More optimization for `MessagingGatewaySupport` * Upgrade to Reactor 3.1 M3 * Document WebFlux-based components Minor Doc Polishing --- build.gradle | 13 +- .../gateway/MessagingGatewaySupport.java | 240 +++++++-- .../config/HttpInboundEndpointParser.java | 4 +- ...tpIntegrationConfigurationInitializer.java | 38 +- .../IntegrationGraphControllerParser.java | 4 +- ...raphControllerRegistrarImportSelector.java | 5 +- .../http/dsl/BaseHttpInboundEndpointSpec.java | 396 +-------------- .../integration/http/dsl/Http.java | 23 + .../dsl/HttpInboundEndpointSupportSpec.java | 444 +++++++++++++++++ .../dsl/ReactiveHttpInboundEndpointSpec.java | 55 +++ .../http/inbound/BaseHttpInboundEndpoint.java | 330 +++++++++++++ .../HttpRequestHandlingEndpointSupport.java | 299 ++---------- .../HttpRequestHandlingMessagingGateway.java | 2 +- .../IntegrationHandlerResultHandler.java | 58 +++ ...tegrationRequestMappingHandlerMapping.java | 14 +- .../inbound/ReactiveHttpInboundEndpoint.java | 461 ++++++++++++++++++ ...tegrationRequestMappingHandlerMapping.java | 170 +++++++ .../http/inbound/RequestMapping.java | 12 +- .../http/support/HttpContextUtils.java | 59 ++- .../integration/http/dsl/HttpDslTests.java | 76 ++- ...pRequestHandlingMessagingGatewayTests.java | 11 +- .../inbound/MultipartAsRawByteArrayTests.java | 3 +- .../ReactiveHttpInboundEndpointTests.java | 170 +++++++ .../JdbcLockRegistryLeaderInitiatorTests.java | 10 +- src/reference/asciidoc/http.adoc | 110 ++++- src/reference/asciidoc/whats-new.adoc | 6 +- 26 files changed, 2268 insertions(+), 745 deletions(-) create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java create mode 100644 spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java create mode 100644 spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java diff --git a/build.gradle b/build.gradle index 0199f814d0..021bddb3f5 100644 --- a/build.gradle +++ b/build.gradle @@ -106,7 +106,7 @@ subprojects { subproject -> hibernateVersion = '5.2.10.Final' hsqldbVersion = '2.4.0' h2Version = '1.4.194' - jackson2Version = '2.9.0.pr3' + jackson2Version = '2.9.0.pr4' javaxActivationVersion = '1.1.1' javaxMailVersion = '1.6.0-rc1' jedisVersion = '2.9.0' @@ -124,8 +124,8 @@ subprojects { subproject -> mysqlVersion = '5.1.41' pahoMqttClientVersion = '1.1.1' postgresVersion = '42.0.0' - reactorNettyVersion = '0.7.0.BUILD-SNAPSHOT' - reactorVersion = '3.1.0.BUILD-SNAPSHOT' + reactorNettyVersion = '0.7.0.M1' + reactorVersion = '3.1.0.M3' romeToolsVersion = '1.7.2' servletApiVersion = '3.1.0' slf4jVersion = "1.7.25" @@ -152,7 +152,7 @@ subprojects { subproject -> } jacoco { - toolVersion = "0.7.8" + toolVersion = "0.7.9" } // dependencies that are common across all java projects @@ -396,8 +396,9 @@ project('spring-integration-http') { description = 'Spring Integration HTTP Support' dependencies { compile project(":spring-integration-core") - compile "org.springframework:spring-webmvc:$springVersion" - compile "org.springframework:spring-webflux:$springVersion" + compile "org.springframework:spring-web:$springVersion" + compile ("org.springframework:spring-webmvc:$springVersion", optional) + compile ("org.springframework:spring-webflux:$springVersion", optional) compile ("javax.servlet:javax.servlet-api:$servletApiVersion", provided) compile ("com.rometools:rome:$romeToolsVersion", optional) compile ("io.projectreactor.ipc:reactor-netty:$reactorNettyVersion" , optional) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index cc23adbef6..2fba505acb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -16,35 +16,48 @@ package org.springframework.integration.gateway; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicLong; +import org.reactivestreams.Subscriber; + import org.springframework.core.AttributeAccessor; import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.history.HistoryWritingMessagePostProcessor; import org.springframework.integration.mapping.InboundMessageMapper; +import org.springframework.integration.mapping.MessageMappingException; import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.integration.support.management.MessageSourceMetrics; import org.springframework.integration.support.management.TrackableComponent; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; +import reactor.core.publisher.Mono; + /** * A convenient base class for connecting application code to * {@link MessageChannel}s for sending, receiving, or request-reply operations. @@ -450,10 +463,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint if (requestChannel == null) { throw new MessagingException("No request channel available. Cannot send request message."); } - MessageChannel replyChannel = getReplyChannel(); - if (replyChannel != null && this.replyMessageCorrelator == null) { - this.registerReplyMessageCorrelator(); - } + + registerReplyMessageCorrelatorIfNecessary(); + Object reply = null; Throwable error = null; Message requestMessage = null; @@ -533,6 +545,129 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return reply; } + protected Mono> sendAndReceiveMessageReactive(Object object) { + initializeIfNecessary(); + Assert.notNull(object, "request must not be null"); + MessageChannel requestChannel = getRequestChannel(); + if (requestChannel == null) { + throw new MessagingException("No request channel available. Cannot send request message."); + } + + registerReplyMessageCorrelatorIfNecessary(); + + return doSendAndReceiveMessageReactive(requestChannel, object, false); + } + + @SuppressWarnings("unchecked") + private Mono> doSendAndReceiveMessageReactive(MessageChannel requestChannel, Object object, + boolean error) { + + return Mono.defer(() -> { + Message message; + try { + message = object instanceof Message + ? (Message) object + : this.requestMapper.toMessage(object); + + message = this.historyWritingPostProcessor.postProcessMessage(message); + + } + catch (Exception e) { + throw new MessageMappingException("Cannot map to message: " + object, e); + } + + Object originalReplyChannelHeader = message.getHeaders().getReplyChannel(); + Object originalErrorChannelHeader = message.getHeaders().getErrorChannel(); + + FutureReplyChannel replyChannel = new FutureReplyChannel(); + + Message requestMessage = MutableMessageBuilder.fromMessage(message) + .setReplyChannel(replyChannel) + .setHeader(this.messagingTemplate.getSendTimeoutHeader(), null) + .setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null) + .setErrorChannel(replyChannel) + .build(); + + if (requestChannel instanceof ReactiveStreamsSubscribableChannel) { + ((ReactiveStreamsSubscribableChannel) requestChannel) + .subscribeTo(Mono.just(requestMessage)); + } + else { + long sendTimeout = sendTimeout(requestMessage); + + boolean sent = + sendTimeout >= 0 + ? requestChannel.send(requestMessage, sendTimeout) + : requestChannel.send(requestMessage); + + if (!sent) { + throw new MessageDeliveryException(requestMessage, + "Failed to send message to channel '" + requestChannel + + "' within timeout: " + sendTimeout); + } + } + + return Mono.fromFuture(replyChannel.messageFuture) + .doOnSubscribe(s -> { + if (!error && this.countsEnabled) { + this.messageCount.incrementAndGet(); + } + }) + .>map(replyMessage -> + MessageBuilder.fromMessage(replyMessage) + .setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader) + .setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader) + .build()) + + .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); + }); + } + + private Mono> handleSendError(Message requestMessage, Throwable exception) { + if (logger.isDebugEnabled()) { + logger.debug("failure occurred in gateway sendAndReceiveReactive: " + exception.getMessage()); + } + MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { + ErrorMessage errorMessage = buildErrorMessage(requestMessage, exception); + try { + return doSendAndReceiveMessageReactive(errorChannel, errorMessage, true); + } + catch (Exception errorFlowFailure) { + throw new MessagingException(errorMessage, "failure occurred in error-handling flow", errorFlowFailure); + } + } + else { + // no errorChannel so we'll propagate + throw wrapExceptionIfNecessary(exception, "gateway received checked Exception"); + } + } + + private long sendTimeout(Message requestMessage) { + Long sendTimeout = headerToLong(requestMessage.getHeaders() + .get(this.messagingTemplate.getSendTimeoutHeader())); + return (sendTimeout != null ? sendTimeout : this.messagingTemplate.getSendTimeout()); + } + + private long receiveTimeout(Message requestMessage) { + Long receiveTimeout = headerToLong(requestMessage.getHeaders() + .get(this.messagingTemplate.getReceiveTimeoutHeader())); + return (receiveTimeout != null ? receiveTimeout : this.messagingTemplate.getReceiveTimeout()); + } + + @Nullable + private Long headerToLong(@Nullable Object headerValue) { + if (headerValue instanceof Number) { + return ((Number) headerValue).longValue(); + } + else if (headerValue instanceof String) { + return Long.parseLong((String) headerValue); + } + else { + return null; + } + } + /** * Build an error message for the message and throwable using the configured * {@link ErrorMessageStrategy}. @@ -542,9 +677,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint * @since 4.3.10 */ protected final ErrorMessage buildErrorMessage(Message requestMessage, Throwable throwable) { - ErrorMessage errorMessage = this.errorMessageStrategy.buildErrorMessage(throwable, - getErrorMessageAttributes(requestMessage)); - return errorMessage; + return this.errorMessageStrategy.buildErrorMessage(throwable, getErrorMessageAttributes(requestMessage)); } /** @@ -560,42 +693,60 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } private void rethrow(Throwable t, String description) { - if (t instanceof RuntimeException) { - throw (RuntimeException) t; - } - throw new MessagingException(description, t); + throw wrapExceptionIfNecessary(t, description); } - private void registerReplyMessageCorrelator() { - synchronized (this.replyMessageCorrelatorMonitor) { - if (this.replyMessageCorrelator != null) { - return; + private RuntimeException wrapExceptionIfNecessary(Throwable t, String description) { + if (t instanceof RuntimeException) { + return (RuntimeException) t; + } + else { + return new MessagingException(description, t); + } + } + + protected void registerReplyMessageCorrelatorIfNecessary() { + MessageChannel replyChannel = getReplyChannel(); + if (replyChannel != null && this.replyMessageCorrelator == null) { + boolean shouldStartCorrelator; + synchronized (this.replyMessageCorrelatorMonitor) { + if (this.replyMessageCorrelator != null) { + return; + } + AbstractEndpoint correlator; + BridgeHandler handler = new BridgeHandler(); + if (getBeanFactory() != null) { + handler.setBeanFactory(getBeanFactory()); + } + handler.afterPropertiesSet(); + if (replyChannel instanceof SubscribableChannel) { + correlator = new EventDrivenConsumer((SubscribableChannel) replyChannel, handler); + } + else if (replyChannel instanceof PollableChannel) { + PollingConsumer endpoint = new PollingConsumer((PollableChannel) replyChannel, handler); + endpoint.setBeanFactory(getBeanFactory()); + endpoint.setReceiveTimeout(this.replyTimeout); + endpoint.afterPropertiesSet(); + correlator = endpoint; + } + else if (replyChannel instanceof ReactiveStreamsSubscribableChannel) { + ReactiveStreamsConsumer endpoint = + new ReactiveStreamsConsumer(replyChannel, (Subscriber>) handler); + endpoint.afterPropertiesSet(); + correlator = endpoint; + } + else { + throw new MessagingException("Unsupported 'replyChannel' type [" + replyChannel.getClass() + "]." + + "SubscribableChannel or PollableChannel type are supported."); + } + this.replyMessageCorrelator = correlator; + shouldStartCorrelator = true; } - AbstractEndpoint correlator = null; - BridgeHandler handler = new BridgeHandler(); - if (this.getBeanFactory() != null) { - handler.setBeanFactory(this.getBeanFactory()); + if (shouldStartCorrelator && isRunning()) { + if (isRunning()) { + this.replyMessageCorrelator.start(); + } } - handler.afterPropertiesSet(); - MessageChannel replyChannel = getReplyChannel(); - if (replyChannel instanceof SubscribableChannel) { - correlator = new EventDrivenConsumer((SubscribableChannel) replyChannel, handler); - } - else if (replyChannel instanceof PollableChannel) { - PollingConsumer endpoint = new PollingConsumer((PollableChannel) replyChannel, handler); - endpoint.setBeanFactory(this.getBeanFactory()); - endpoint.setReceiveTimeout(this.replyTimeout); - endpoint.afterPropertiesSet(); - correlator = endpoint; - } - else { - throw new MessagingException("Unsupported 'replyChannel' type [" + replyChannel.getClass() + "]." - + "SubscribableChannel or PollableChannel type are supported."); - } - if (this.isRunning()) { - correlator.start(); - } - this.replyMessageCorrelator = correlator; } } @@ -641,4 +792,15 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } + private static class FutureReplyChannel implements MessageChannel { + + private final CompletableFuture> messageFuture = new CompletableFuture<>(); + + @Override + public boolean send(Message message, long timeout) { + return this.messageFuture.complete(message); + } + + } + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java index a30967e18e..0825c4aa46 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.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. @@ -191,7 +191,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse builder.addPropertyValue("crossOrigin", crossOriginBuilder.getBeanDefinition()); } - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "request-payload-type", "requestPayloadType"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "request-payload-type", "requestPayloadTypeClass"); BeanDefinition statusCodeExpressionDef = IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("status-code-expression", element); diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java index 4668cbae4f..82c9e55178 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -23,16 +23,21 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.integration.config.IntegrationConfigurationInitializer; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.http.inbound.IntegrationHandlerResultHandler; import org.springframework.integration.http.inbound.IntegrationRequestMappingHandlerMapping; +import org.springframework.integration.http.inbound.ReactiveIntegrationRequestMappingHandlerMapping; import org.springframework.integration.http.support.HttpContextUtils; /** * The HTTP Integration infrastructure {@code beanFactory} initializer. * * @author Artem Bilan + * * @since 4.0 */ public class HttpIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { @@ -42,7 +47,8 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi @Override public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException { if (beanFactory instanceof BeanDefinitionRegistry) { - this.registerRequestMappingHandlerMappingIfNecessary((BeanDefinitionRegistry) beanFactory); + registerRequestMappingHandlerMappingIfNecessary((BeanDefinitionRegistry) beanFactory); + registerReactiveRequestMappingHandlerMappingIfNecessary((BeanDefinitionRegistry) beanFactory); } else { logger.warn("'IntegrationRequestMappingHandlerMapping' isn't registered because 'beanFactory'" + @@ -61,7 +67,7 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi * the HTTP server components. */ private void registerRequestMappingHandlerMappingIfNecessary(BeanDefinitionRegistry registry) { - if (HttpContextUtils.SERVLET_PRESENT && + if (HttpContextUtils.WEB_MVC_PRESENT && !registry.containsBeanDefinition(HttpContextUtils.HANDLER_MAPPING_BEAN_NAME)) { BeanDefinitionBuilder requestMappingBuilder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationRequestMappingHandlerMapping.class); @@ -72,4 +78,30 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi } } + /** + * Registers a {@link ReactiveIntegrationRequestMappingHandlerMapping} + * which could also be overridden by the user by simply registering + * a {@link ReactiveIntegrationRequestMappingHandlerMapping} {@code } with 'id' + * {@link HttpContextUtils#REACTIVE_HANDLER_MAPPING_BEAN_NAME}. + *

+ * In addition, checks if the {@code org.springframework.web.reactive.result.method.RequestMappingInfo} + * class is present on the classpath. + * When Spring Integration HTTP is used only as an HTTP client, there is no reason to use and register + * the HTTP server components. + */ + private void registerReactiveRequestMappingHandlerMappingIfNecessary(BeanDefinitionRegistry registry) { + if (HttpContextUtils.WEB_FLUX_PRESENT && + !registry.containsBeanDefinition(HttpContextUtils.REACTIVE_HANDLER_MAPPING_BEAN_NAME)) { + BeanDefinitionBuilder requestMappingBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ReactiveIntegrationRequestMappingHandlerMapping.class); + requestMappingBuilder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); + requestMappingBuilder.addPropertyValue(IntegrationNamespaceUtils.ORDER, 0); + registry.registerBeanDefinition(HttpContextUtils.REACTIVE_HANDLER_MAPPING_BEAN_NAME, + requestMappingBuilder.getBeanDefinition()); + + BeanDefinitionReaderUtils.registerWithGeneratedName( + new RootBeanDefinition(IntegrationHandlerResultHandler.class), registry); + } + } + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerParser.java index 47042ffdb8..a78af6ea6a 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -40,7 +40,7 @@ public class IntegrationGraphControllerParser implements BeanDefinitionParser { @Override public BeanDefinition parse(final Element element, ParserContext parserContext) { - if (HttpContextUtils.SERVLET_PRESENT) { + if (HttpContextUtils.WEB_MVC_PRESENT) { this.graphControllerRegistrar.registerBeanDefinitions( new StandardAnnotationMetadata(IntegrationGraphControllerParser.class) { diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerRegistrarImportSelector.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerRegistrarImportSelector.java index 5d5cc0ca99..0509cae676 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerRegistrarImportSelector.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerRegistrarImportSelector.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,6 +25,7 @@ import org.springframework.integration.http.support.HttpContextUtils; /** * @author Artem Bilan + * * @since 4.3 */ class IntegrationGraphControllerRegistrarImportSelector implements ImportSelector { @@ -33,7 +34,7 @@ class IntegrationGraphControllerRegistrarImportSelector implements ImportSelecto @Override public String[] selectImports(AnnotationMetadata importingClassMetadata) { - if (HttpContextUtils.SERVLET_PRESENT) { + if (HttpContextUtils.WEB_MVC_PRESENT) { return new String[] { IntegrationGraphControllerRegistrar.class.getName() }; } else { diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpInboundEndpointSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpInboundEndpointSpec.java index 0c9235b549..d04c096b62 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpInboundEndpointSpec.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/BaseHttpInboundEndpointSpec.java @@ -17,27 +17,10 @@ package org.springframework.integration.http.dsl; import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; -import java.util.function.Consumer; -import java.util.function.Function; -import org.springframework.expression.Expression; -import org.springframework.http.HttpEntity; -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.MessagingGatewaySpec; -import org.springframework.integration.expression.FunctionExpression; -import org.springframework.integration.http.inbound.CrossOrigin; import org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport; -import org.springframework.integration.http.inbound.RequestMapping; -import org.springframework.integration.http.support.DefaultHttpHeaderMapper; -import org.springframework.integration.mapping.HeaderMapper; -import org.springframework.util.Assert; -import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.multipart.MultipartResolver; /** @@ -52,126 +35,10 @@ import org.springframework.web.multipart.MultipartResolver; */ public abstract class BaseHttpInboundEndpointSpec, E extends HttpRequestHandlingEndpointSupport> - extends MessagingGatewaySpec implements ComponentsRegistration { - - private final RequestMapping requestMapping = new RequestMapping(); - - private final Map headerExpressions = new HashMap<>(); - - private final HeaderMapper headerMapper = DefaultHttpHeaderMapper.inboundMapper(); - - private HeaderMapper explicitHeaderMapper; + extends HttpInboundEndpointSupportSpec { BaseHttpInboundEndpointSpec(E endpoint, String... path) { - super(endpoint); - this.requestMapping.setPathPatterns(path); - this.target.setRequestMapping(this.requestMapping); - this.target.setHeaderExpressions(this.headerExpressions); - this.target.setHeaderMapper(this.headerMapper); - } - - /** - * Provide a {@link Consumer} for configuring {@link RequestMapping} via {@link RequestMappingSpec} - * @param requestMapping the {@link Consumer} to configure {@link RequestMappingSpec}. - * @return the spec - * @see RequestMapping - */ - public S requestMapping(Consumer requestMapping) { - requestMapping.accept(new RequestMappingSpec(this.requestMapping)); - return _this(); - } - - /** - * Provide a {@link Consumer} for configuring {@link CrossOrigin} via {@link CrossOriginSpec} - * @param crossOrigin the {@link Consumer} to configure {@link CrossOriginSpec}. - * @return the spec - * @see CrossOrigin - */ - public S crossOrigin(Consumer crossOrigin) { - CrossOriginSpec originSpec = new CrossOriginSpec(); - crossOrigin.accept(originSpec); - this.target.setCrossOrigin(originSpec.crossOrigin); - return _this(); - } - - /** - * Specify a SpEL expression to evaluate in order to generate the Message payload. - * @param payloadExpression The payload expression. - * @return the spec - * @see HttpRequestHandlingEndpointSupport#setPayloadExpression(Expression) - */ - public S payloadExpression(String payloadExpression) { - return payloadExpression(PARSER.parseExpression(payloadExpression)); - } - - /** - * Specify a SpEL expression to evaluate in order to generate the Message payload. - * @param payloadExpression The payload expression. - * @return the spec - * @see HttpRequestHandlingEndpointSupport#setPayloadExpression(Expression) - */ - public S payloadExpression(Expression payloadExpression) { - this.target.setPayloadExpression(payloadExpression); - return _this(); - } - - /** - * Specify a {@link Function} to evaluate in order to generate the Message payload. - * @param payloadFunction The payload {@link Function}. - * @param

the expected HTTP request body type. - * @return the spec - * @see HttpRequestHandlingEndpointSupport#setPayloadExpression(Expression) - */ - public

S payloadFunction(Function, ?> payloadFunction) { - return payloadExpression(new FunctionExpression<>(payloadFunction)); - } - - /** - * Specify a Map of SpEL expressions to evaluate in order to generate the Message headers. - * @param headerExpressions The {@link Map} of SpEL expressions for headers. - * @return the spec - * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) - */ - public S headerExpressions(Map headerExpressions) { - Assert.notNull(headerExpressions, "'headerExpressions' must not be null"); - this.headerExpressions.clear(); - this.headerExpressions.putAll(headerExpressions); - return _this(); - } - - /** - * Specify SpEL expression for provided header to populate. - * @param header the header name to populate. - * @param expression the SpEL expression for the header. - * @return the spec - * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) - */ - public S headerExpression(String header, String expression) { - return headerExpression(header, PARSER.parseExpression(expression)); - } - - /** - * Specify SpEL expression for provided header to populate. - * @param header the header name to populate. - * @param expression the SpEL expression for the header. - * @return the spec - * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) - */ - public S headerExpression(String header, Expression expression) { - this.headerExpressions.put(header, expression); - return _this(); - } - - /** - * Specify a {@link Function} for provided header to populate. - * @param header the header name to add. - * @param headerFunction the function to evaluate the header value against {@link HttpEntity}. - * @param

the expected HTTP body type. - * @return the current Spec. - * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) - */ - public

S headerFunction(String header, Function, ?> headerFunction) { - return headerExpression(header, new FunctionExpression<>(headerFunction)); + super(endpoint, path); } /** @@ -195,70 +62,6 @@ public abstract class BaseHttpInboundEndpointSpec headerMapper) { - this.target.setHeaderMapper(headerMapper); - this.explicitHeaderMapper = headerMapper; - return _this(); - } - - /** - * Provide the pattern array for request headers to map. - * @param patterns the patterns for request headers to map. - * @return the current Spec. - * @see DefaultHttpHeaderMapper#setOutboundHeaderNames(String[]) - */ - public S mappedRequestHeaders(String... patterns) { - Assert.isNull(this.explicitHeaderMapper, - "The 'mappedRequestHeaders' must be specified on the provided 'headerMapper': " - + this.explicitHeaderMapper); - ((DefaultHttpHeaderMapper) this.headerMapper).setInboundHeaderNames(patterns); - return _this(); - } - - /** - * Provide the pattern array for response headers to map. - * @param patterns the patterns for response headers to map. - * @return the current Spec. - * @see DefaultHttpHeaderMapper#setInboundHeaderNames(String[]) - */ - public S mappedResponseHeaders(String... patterns) { - Assert.isNull(this.explicitHeaderMapper, - "The 'mappedRequestHeaders' must be specified on the provided 'headerMapper': " - + this.explicitHeaderMapper); - ((DefaultHttpHeaderMapper) this.headerMapper).setOutboundHeaderNames(patterns); - return _this(); - } - - /** - * Specify the type of payload to be generated when the inbound HTTP request content is read by the - * {@link HttpMessageConverter}s. - * By default this value is null which means at runtime any "text" Content-Type will - * result in String while all others default to byte[].class. - * @param requestPayloadType The payload type. - * @return the current Spec. - */ - public S requestPayloadType(Class requestPayloadType) { - this.target.setRequestPayloadType(requestPayloadType); - return _this(); - } - - /** - * Specify whether only the reply Message's payload should be passed in the response. - * If this is set to {@code false}, the entire Message will be used to generate the response. - * The default is {@code true}. - * @param extractReplyPayload true to extract the reply payload. - * @return the current Spec. - */ - public S extractReplyPayload(boolean extractReplyPayload) { - this.target.setExtractReplyPayload(extractReplyPayload); - return _this(); - } - /** * Specify the {@link MultipartResolver} to use when checking requests. * @param multipartResolver The multipart resolver. @@ -269,199 +72,4 @@ public abstract class BaseHttpInboundEndpointSpec statusCodeFunction) { - return statusCodeExpression(new FunctionExpression<>(statusCodeFunction)); - } - - @Override - public Map getComponentsToRegister() { - HeaderMapper headerMapperToRegister = - (this.explicitHeaderMapper != null ? this.explicitHeaderMapper : this.headerMapper); - return Collections.singletonMap(headerMapperToRegister, null); - } - - /** - * A fluent API for the {@link RequestMapping}. - */ - public static final class RequestMappingSpec { - - private final RequestMapping requestMapping; - - RequestMappingSpec(RequestMapping requestMapping) { - this.requestMapping = requestMapping; - } - - /** - * The HTTP request methods to map to, narrowing the primary mapping: - * GET, POST, HEAD, OPTIONS, PUT, PATCH, DELETE, TRACE. - * @param supportedMethods the {@link HttpMethod}s to use. - * @return the spec - */ - public RequestMappingSpec methods(HttpMethod... supportedMethods) { - this.requestMapping.setMethods(supportedMethods); - return this; - } - - /** - * The parameters of the mapped request, narrowing the primary mapping. - * @param params the request params to map to. - * @return the spec - */ - public RequestMappingSpec params(String... params) { - this.requestMapping.setParams(params); - return this; - } - - /** - * The headers of the mapped request, narrowing the primary mapping. - * @param headers the request headers to map to. - * @return the spec - */ - public RequestMappingSpec headers(String... headers) { - this.requestMapping.setHeaders(headers); - return this; - } - - /** - * The consumable media types of the mapped request, narrowing the primary mapping. - * @param consumes the the media types for {@code Content-Type} header. - * @return the spec - */ - public RequestMappingSpec consumes(String... consumes) { - this.requestMapping.setConsumes(consumes); - return this; - } - - /** - * The producible media types of the mapped request, narrowing the primary mapping. - * @param produces the the media types for {@code Accept} header. - * @return the spec - */ - public RequestMappingSpec produces(String... produces) { - this.requestMapping.setProduces(produces); - return this; - } - - } - - /** - * A fluent API for the {@link CrossOrigin}. - */ - public static final class CrossOriginSpec { - - private final CrossOrigin crossOrigin = new CrossOrigin(); - - CrossOriginSpec() { - super(); - } - - /** - * List of allowed origins, e.g. {@code "http://domain1.com"}. - *

These values are placed in the {@code Access-Control-Allow-Origin} - * header of both the pre-flight response and the actual response. - * {@code "*"} means that all origins are allowed. - *

If undefined, all origins are allowed. - * @param origin the list of allowed origins. - * @return the spec - */ - public CrossOriginSpec origin(String... origin) { - this.crossOrigin.setOrigin(origin); - return this; - } - - /** - * List of request headers that can be used during the actual request. - *

This property controls the value of the pre-flight response's - * {@code Access-Control-Allow-Headers} header. - * {@code "*"} means that all headers requested by the client are allowed. - * @param allowedHeaders the list of request headers. - * @return the spec - */ - public CrossOriginSpec allowedHeaders(String... allowedHeaders) { - this.crossOrigin.setAllowedHeaders(allowedHeaders); - return this; - } - - /** - * List of response headers that the user-agent will allow the client to access. - *

This property controls the value of actual response's - * {@code Access-Control-Expose-Headers} header. - * @param exposedHeaders the list of response headers. - * @return the spec - */ - public CrossOriginSpec exposedHeaders(String... exposedHeaders) { - this.crossOrigin.setExposedHeaders(exposedHeaders); - return this; - } - - /** - * List of supported HTTP request methods, e.g. - * {@code "{RequestMethod.GET, RequestMethod.POST}"}. - *

Methods specified here override those specified via {@code RequestMapping}. - * @param method the list of supported HTTP request methods - * @return the spec - */ - public CrossOriginSpec method(RequestMethod... method) { - this.crossOrigin.setMethod(method); - return this; - } - - /** - * Whether the browser should include any cookies associated with the - * domain of the request being annotated. - *

Set to {@code "false"} if such cookies should not included. - * @param allowCredentials the {@code boolean} flag to include - * {@code Access-Control-Allow-Credentials=true} in pre-flight response or not - * @return the spec - */ - public CrossOriginSpec allowCredentials(Boolean allowCredentials) { - this.crossOrigin.setAllowCredentials(allowCredentials); - return this; - } - - /** - * The maximum age (in seconds) of the cache duration for pre-flight responses. - *

This property controls the value of the {@code Access-Control-Max-Age} - * header in the pre-flight response. - * @param maxAge the maximum age (in seconds) of the cache duration for pre-flight responses. - * @return the spec - */ - public CrossOriginSpec maxAge(long maxAge) { - this.crossOrigin.setMaxAge(maxAge); - return this; - } - - } - } 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 14ebba59cd..80c9addbdb 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 @@ -24,6 +24,7 @@ import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.http.inbound.HttpRequestHandlingController; import org.springframework.integration.http.inbound.HttpRequestHandlingMessagingGateway; +import org.springframework.integration.http.inbound.ReactiveHttpInboundEndpoint; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -467,6 +468,28 @@ public final class Http { return new HttpRequestHandlerEndpointSpec(new HttpRequestHandlingMessagingGateway(), path); } + + /** + * Create an {@link ReactiveHttpInboundEndpointSpec} builder for one-way reactive adapter + * based on the provided {@code path} array for mapping. + * @param path the path mapping URIs (e.g. "/myPath.do"). + * @return the ReactiveHttpInboundEndpointSpec instance + */ + public static ReactiveHttpInboundEndpointSpec inboundReactiveChannelAdapter(String... path) { + ReactiveHttpInboundEndpoint httpInboundChannelAdapter = new ReactiveHttpInboundEndpoint(false); + return new ReactiveHttpInboundEndpointSpec(httpInboundChannelAdapter, path); + } + + /** + * Create an {@link ReactiveHttpInboundEndpointSpec} builder for request-reply reactive gateway + * based on the provided {@code path} array for mapping. + * @param path the path mapping URIs (e.g. "/myPath.do"). + * @return the ReactiveHttpInboundEndpointSpec instance + */ + public static ReactiveHttpInboundEndpointSpec inboundReactiveGateway(String... path) { + return new ReactiveHttpInboundEndpointSpec(new ReactiveHttpInboundEndpoint(), path); + } + private Http() { super(); } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java new file mode 100644 index 0000000000..682e213cc1 --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/HttpInboundEndpointSupportSpec.java @@ -0,0 +1,444 @@ +/* + * 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.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.springframework.core.ResolvableType; +import org.springframework.expression.Expression; +import org.springframework.http.HttpEntity; +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.MessagingGatewaySpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.http.inbound.BaseHttpInboundEndpoint; +import org.springframework.integration.http.inbound.CrossOrigin; +import org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport; +import org.springframework.integration.http.inbound.RequestMapping; +import org.springframework.integration.http.support.DefaultHttpHeaderMapper; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.util.Assert; +import org.springframework.web.bind.annotation.RequestMethod; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +public abstract class HttpInboundEndpointSupportSpec, E extends BaseHttpInboundEndpoint> + extends MessagingGatewaySpec + implements ComponentsRegistration { + + protected final RequestMapping requestMapping = new RequestMapping(); + + protected final Map headerExpressions = new HashMap<>(); + + protected final HeaderMapper headerMapper = DefaultHttpHeaderMapper.inboundMapper(); + + private HeaderMapper explicitHeaderMapper; + + HttpInboundEndpointSupportSpec(E gateway, String... path) { + super(gateway); + this.requestMapping.setPathPatterns(path); + this.target.setRequestMapping(this.requestMapping); + this.target.setHeaderExpressions(this.headerExpressions); + this.target.setHeaderMapper(this.headerMapper); + } + + /** + * Provide a {@link Consumer} for configuring {@link RequestMapping} via {@link RequestMappingSpec} + * @param requestMapping the {@link Consumer} to configure {@link RequestMappingSpec}. + * @return the spec + * @see RequestMapping + */ + public S requestMapping(Consumer requestMapping) { + requestMapping.accept(new RequestMappingSpec(this.requestMapping)); + return _this(); + } + + /** + * Provide a {@link Consumer} for configuring {@link CrossOrigin} via {@link CrossOriginSpec} + * @param crossOrigin the {@link Consumer} to configure {@link CrossOriginSpec}. + * @return the spec + * @see CrossOrigin + */ + public S crossOrigin(Consumer crossOrigin) { + CrossOriginSpec originSpec = new CrossOriginSpec(); + crossOrigin.accept(originSpec); + this.target.setCrossOrigin(originSpec.crossOrigin); + return _this(); + } + + /** + * Specify a SpEL expression to evaluate in order to generate the Message payload. + * @param payloadExpression The payload expression. + * @return the spec + * @see HttpRequestHandlingEndpointSupport#setPayloadExpression(Expression) + */ + public S payloadExpression(String payloadExpression) { + return payloadExpression(PARSER.parseExpression(payloadExpression)); + } + + /** + * Specify a SpEL expression to evaluate in order to generate the Message payload. + * @param payloadExpression The payload expression. + * @return the spec + * @see HttpRequestHandlingEndpointSupport#setPayloadExpression(Expression) + */ + public S payloadExpression(Expression payloadExpression) { + this.target.setPayloadExpression(payloadExpression); + return _this(); + } + + /** + * Specify a {@link Function} to evaluate in order to generate the Message payload. + * @param payloadFunction The payload {@link Function}. + * @param

the expected HTTP request body type. + * @return the spec + * @see HttpRequestHandlingEndpointSupport#setPayloadExpression(Expression) + */ + public

S payloadFunction(Function, ?> payloadFunction) { + return payloadExpression(new FunctionExpression<>(payloadFunction)); + } + + /** + * Specify a Map of SpEL expressions to evaluate in order to generate the Message headers. + * @param headerExpressions The {@link Map} of SpEL expressions for headers. + * @return the spec + * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) + */ + public S headerExpressions(Map headerExpressions) { + Assert.notNull(headerExpressions, "'headerExpressions' must not be null"); + this.headerExpressions.clear(); + this.headerExpressions.putAll(headerExpressions); + return _this(); + } + + /** + * Specify SpEL expression for provided header to populate. + * @param header the header name to populate. + * @param expression the SpEL expression for the header. + * @return the spec + * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) + */ + public S headerExpression(String header, String expression) { + return headerExpression(header, PARSER.parseExpression(expression)); + } + + /** + * Specify SpEL expression for provided header to populate. + * @param header the header name to populate. + * @param expression the SpEL expression for the header. + * @return the spec + * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) + */ + public S headerExpression(String header, Expression expression) { + this.headerExpressions.put(header, expression); + return _this(); + } + + /** + * Specify a {@link Function} for provided header to populate. + * @param header the header name to add. + * @param headerFunction the function to evaluate the header value against {@link HttpEntity}. + * @param

the expected HTTP body type. + * @return the current Spec. + * @see HttpRequestHandlingEndpointSupport#setHeaderExpressions(Map) + */ + public

S headerFunction(String header, Function, ?> headerFunction) { + return headerExpression(header, new FunctionExpression<>(headerFunction)); + } + + /** + * Set the {@link HeaderMapper} to use when mapping between HTTP headers and MessageHeaders. + * @param headerMapper The header mapper. + * @return the current Spec. + */ + public S headerMapper(HeaderMapper headerMapper) { + this.target.setHeaderMapper(headerMapper); + this.explicitHeaderMapper = headerMapper; + return _this(); + } + + /** + * Provide the pattern array for request headers to map. + * @param patterns the patterns for request headers to map. + * @return the current Spec. + * @see DefaultHttpHeaderMapper#setOutboundHeaderNames(String[]) + */ + public S mappedRequestHeaders(String... patterns) { + Assert.isNull(this.explicitHeaderMapper, + "The 'mappedRequestHeaders' must be specified on the provided 'headerMapper': " + + this.explicitHeaderMapper); + ((DefaultHttpHeaderMapper) this.headerMapper).setInboundHeaderNames(patterns); + return _this(); + } + + /** + * Provide the pattern array for response headers to map. + * @param patterns the patterns for response headers to map. + * @return the current Spec. + * @see DefaultHttpHeaderMapper#setInboundHeaderNames(String[]) + */ + public S mappedResponseHeaders(String... patterns) { + Assert.isNull(this.explicitHeaderMapper, + "The 'mappedRequestHeaders' must be specified on the provided 'headerMapper': " + + this.explicitHeaderMapper); + ((DefaultHttpHeaderMapper) this.headerMapper).setOutboundHeaderNames(patterns); + return _this(); + } + + /** + * Specify the type of payload to be generated when the inbound HTTP request content is read by the + * {@link HttpMessageConverter}s. + * By default this value is null which means at runtime any "text" Content-Type will + * result in String while all others default to byte[].class. + * @param requestPayloadType The payload type. + * @return the current Spec. + */ + public S requestPayloadType(Class requestPayloadType) { + this.target.setRequestPayloadTypeClass(requestPayloadType); + return _this(); + } + + /** + * Specify the type of payload to be generated when the inbound HTTP request content is read by the + * {@link HttpMessageConverter}s. + * By default this value is null which means at runtime any "text" Content-Type will + * result in String while all others default to byte[].class. + * @param requestPayloadType The payload type. + * @return the current Spec. + */ + public S requestPayloadType(ResolvableType requestPayloadType) { + this.target.setRequestPayloadType(requestPayloadType); + return _this(); + } + + /** + * Specify whether only the reply Message's payload should be passed in the response. + * If this is set to {@code false}, the entire Message will be used to generate the response. + * The default is {@code true}. + * @param extractReplyPayload true to extract the reply payload. + * @return the current Spec. + */ + public S extractReplyPayload(boolean extractReplyPayload) { + this.target.setExtractReplyPayload(extractReplyPayload); + return _this(); + } + + /** + * Specify the {@link Expression} to resolve a status code for Response to override + * the default '200 OK' or '500 Internal Server Error' for a timeout. + * @param statusCodeExpression The status code Expression. + * @return the current Spec. + * @see HttpRequestHandlingEndpointSupport#setStatusCodeExpression(Expression) + */ + public S statusCodeExpression(String statusCodeExpression) { + this.target.setStatusCodeExpressionString(statusCodeExpression); + return _this(); + } + + /** + * Specify the {@link Expression} to resolve a status code for Response to override + * the default '200 OK' or '500 Internal Server Error' for a timeout. + * @param statusCodeExpression The status code Expression. + * @return the current Spec. + * @see HttpRequestHandlingEndpointSupport#setStatusCodeExpression(Expression) + */ + public S statusCodeExpression(Expression statusCodeExpression) { + this.target.setStatusCodeExpression(statusCodeExpression); + return _this(); + } + + /** + * Specify the {@link Function} to resolve a status code for Response to override + * the default '200 OK' or '500 Internal Server Error' for a timeout. + * @param statusCodeFunction The status code {@link Function}. + * @return the current Spec. + * @see HttpRequestHandlingEndpointSupport#setStatusCodeExpression(Expression) + */ + public S statusCodeFunction(Function statusCodeFunction) { + return statusCodeExpression(new FunctionExpression<>(statusCodeFunction)); + } + + @Override + public Map getComponentsToRegister() { + HeaderMapper headerMapperToRegister = + (this.explicitHeaderMapper != null ? this.explicitHeaderMapper : this.headerMapper); + return Collections.singletonMap(headerMapperToRegister, null); + } + + /** + * A fluent API for the {@link RequestMapping}. + */ + public static final class RequestMappingSpec { + + private final RequestMapping requestMapping; + + RequestMappingSpec(RequestMapping requestMapping) { + this.requestMapping = requestMapping; + } + + /** + * The HTTP request methods to map to, narrowing the primary mapping: + * GET, POST, HEAD, OPTIONS, PUT, PATCH, DELETE, TRACE. + * @param supportedMethods the {@link HttpMethod}s to use. + * @return the spec + */ + public RequestMappingSpec methods(HttpMethod... supportedMethods) { + this.requestMapping.setMethods(supportedMethods); + return this; + } + + /** + * The parameters of the mapped request, narrowing the primary mapping. + * @param params the request params to map to. + * @return the spec + */ + public RequestMappingSpec params(String... params) { + this.requestMapping.setParams(params); + return this; + } + + /** + * The headers of the mapped request, narrowing the primary mapping. + * @param headers the request headers to map to. + * @return the spec + */ + public RequestMappingSpec headers(String... headers) { + this.requestMapping.setHeaders(headers); + return this; + } + + /** + * The consumable media types of the mapped request, narrowing the primary mapping. + * @param consumes the the media types for {@code Content-Type} header. + * @return the spec + */ + public RequestMappingSpec consumes(String... consumes) { + this.requestMapping.setConsumes(consumes); + return this; + } + + /** + * The producible media types of the mapped request, narrowing the primary mapping. + * @param produces the the media types for {@code Accept} header. + * @return the spec + */ + public RequestMappingSpec produces(String... produces) { + this.requestMapping.setProduces(produces); + return this; + } + + } + + /** + * A fluent API for the {@link CrossOrigin}. + */ + public static final class CrossOriginSpec { + + private final CrossOrigin crossOrigin = new CrossOrigin(); + + CrossOriginSpec() { + super(); + } + + /** + * List of allowed origins, e.g. {@code "http://domain1.com"}. + *

These values are placed in the {@code Access-Control-Allow-Origin} + * header of both the pre-flight response and the actual response. + * {@code "*"} means that all origins are allowed. + *

If undefined, all origins are allowed. + * @param origin the list of allowed origins. + * @return the spec + */ + public CrossOriginSpec origin(String... origin) { + this.crossOrigin.setOrigin(origin); + return this; + } + + /** + * List of request headers that can be used during the actual request. + *

This property controls the value of the pre-flight response's + * {@code Access-Control-Allow-Headers} header. + * {@code "*"} means that all headers requested by the client are allowed. + * @param allowedHeaders the list of request headers. + * @return the spec + */ + public CrossOriginSpec allowedHeaders(String... allowedHeaders) { + this.crossOrigin.setAllowedHeaders(allowedHeaders); + return this; + } + + /** + * List of response headers that the user-agent will allow the client to access. + *

This property controls the value of actual response's + * {@code Access-Control-Expose-Headers} header. + * @param exposedHeaders the list of response headers. + * @return the spec + */ + public CrossOriginSpec exposedHeaders(String... exposedHeaders) { + this.crossOrigin.setExposedHeaders(exposedHeaders); + return this; + } + + /** + * List of supported HTTP request methods, e.g. + * {@code "{RequestMethod.GET, RequestMethod.POST}"}. + *

Methods specified here override those specified via {@code RequestMapping}. + * @param method the list of supported HTTP request methods + * @return the spec + */ + public CrossOriginSpec method(RequestMethod... method) { + this.crossOrigin.setMethod(method); + return this; + } + + /** + * Whether the browser should include any cookies associated with the + * domain of the request being annotated. + *

Set to {@code "false"} if such cookies should not included. + * @param allowCredentials the {@code boolean} flag to include + * {@code Access-Control-Allow-Credentials=true} in pre-flight response or not + * @return the spec + */ + public CrossOriginSpec allowCredentials(Boolean allowCredentials) { + this.crossOrigin.setAllowCredentials(allowCredentials); + return this; + } + + /** + * The maximum age (in seconds) of the cache duration for pre-flight responses. + *

This property controls the value of the {@code Access-Control-Max-Age} + * header in the pre-flight response. + * @param maxAge the maximum age (in seconds) of the cache duration for pre-flight responses. + * @return the spec + */ + public CrossOriginSpec maxAge(long maxAge) { + this.crossOrigin.setMaxAge(maxAge); + return this; + } + + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java new file mode 100644 index 0000000000..b996cf6e17 --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/dsl/ReactiveHttpInboundEndpointSpec.java @@ -0,0 +1,55 @@ +/* + * 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 org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.integration.http.inbound.ReactiveHttpInboundEndpoint; +import org.springframework.web.reactive.accept.RequestedContentTypeResolver; + +/** + * The {@link HttpInboundEndpointSupportSpec} implementation for the {@link ReactiveHttpInboundEndpoint}. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class ReactiveHttpInboundEndpointSpec + extends HttpInboundEndpointSupportSpec { + + ReactiveHttpInboundEndpointSpec(ReactiveHttpInboundEndpoint gateway, String... path) { + super(gateway, path); + } + + public ReactiveHttpInboundEndpointSpec codecConfigurer(ServerCodecConfigurer codecConfigurer) { + this.target.setCodecConfigurer(codecConfigurer); + return this; + } + + public ReactiveHttpInboundEndpointSpec requestedContentTypeResolver( + RequestedContentTypeResolver requestedContentTypeResolver) { + + this.target.setRequestedContentTypeResolver(requestedContentTypeResolver); + return this; + } + + public ReactiveHttpInboundEndpointSpec reactiveAdapterRegistry(ReactiveAdapterRegistry adapterRegistry) { + this.target.setReactiveAdapterRegistry(adapterRegistry); + return this; + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java new file mode 100644 index 0000000000..c31b07abe6 --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java @@ -0,0 +1,330 @@ +/* + * 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.inbound; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.core.ResolvableType; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpRequest; +import org.springframework.http.HttpStatus; +import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.http.support.DefaultHttpHeaderMapper; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.CollectionUtils; + +/** + * The {@link MessagingGatewaySupport} extension for HTTP Inbound endpoints + * with basic properties. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class BaseHttpInboundEndpoint extends MessagingGatewaySupport implements OrderlyShutdownCapable { + + protected static final boolean jaxb2Present = ClassUtils.isPresent("javax.xml.bind.Binder", + HttpRequestHandlingEndpointSupport.class.getClassLoader()); + + protected static final boolean romeToolsPresent = ClassUtils.isPresent("com.rometools.rome.feed.atom.Feed", + HttpRequestHandlingEndpointSupport.class.getClassLoader()); + + protected static final List nonReadableBodyHttpMethods = + Arrays.asList(HttpMethod.GET, HttpMethod.HEAD, HttpMethod.OPTIONS); + + protected final boolean expectReply; + + protected final AtomicInteger activeCount = new AtomicInteger(); + + private volatile ResolvableType requestPayloadType = null; + + private volatile HeaderMapper headerMapper = DefaultHttpHeaderMapper.inboundMapper(); + + private volatile boolean extractReplyPayload = true; + + private volatile Expression statusCodeExpression; + + private volatile EvaluationContext evaluationContext; + + private volatile RequestMapping requestMapping = new RequestMapping(); + + private volatile Expression payloadExpression; + + private volatile Map headerExpressions; + + private volatile CrossOrigin crossOrigin; + + public BaseHttpInboundEndpoint(boolean expectReply) { + super(expectReply); + this.expectReply = expectReply; + } + + /** + * Specifies a SpEL expression to evaluate in order to generate the Message payload. + * The EvaluationContext will be populated with an HttpEntity instance as the root object, + * and it may contain variables: + *

    + *
  • #pathVariables
    • + *
  • #requestParams
    • + *
  • #requestAttributes
    • + *
  • #requestHeaders
    • + *
  • #matrixVariables
    • + *
  • #cookies + *
+ * @param payloadExpression The payload expression. + */ + public void setPayloadExpression(Expression payloadExpression) { + this.payloadExpression = payloadExpression; + } + + /** + * Specifies a Map of SpEL expressions to evaluate in order to generate the Message headers. + * The keys in the map will be used as the header names. When evaluating the expression, + * the EvaluationContext will be populated with an HttpEntity instance as the root object, + * and it may contain variables: + *
    + *
  • #pathVariables
    • + *
  • #requestParams
    • + *
  • #requestAttributes
    • + *
  • #requestHeaders
    • + *
  • #matrixVariables
    • + *
  • #cookies + *
+ * @param headerExpressions The header expressions. + */ + public void setHeaderExpressions(Map headerExpressions) { + this.headerExpressions = headerExpressions; + } + + /** + * Set the {@link RequestMapping} which allows you to specify a flexible RESTFul-mapping for this endpoint. + * @param requestMapping The request mapping. + */ + public void setRequestMapping(RequestMapping requestMapping) { + Assert.notNull(requestMapping, "requestMapping must not be null"); + this.requestMapping = requestMapping; + } + + public final RequestMapping getRequestMapping() { + return this.requestMapping; + } + + /** + * Set the {@link CrossOrigin} to permit cross origin requests for this endpoint. + * @param crossOrigin the CrossOrigin config. + * @since 4.2 + */ + public void setCrossOrigin(CrossOrigin crossOrigin) { + this.crossOrigin = crossOrigin; + } + + public CrossOrigin getCrossOrigin() { + return this.crossOrigin; + } + + protected Expression getPayloadExpression() { + return this.payloadExpression; + } + + protected Map getHeaderExpressions() { + return this.headerExpressions; + } + + /** + * @return Whether to expect a reply. + */ + protected boolean isExpectReply() { + return this.expectReply; + } + + /** + * Set the {@link HeaderMapper} to use when mapping between HTTP headers and MessageHeaders. + * @param headerMapper The header mapper. + */ + public void setHeaderMapper(HeaderMapper headerMapper) { + Assert.notNull(headerMapper, "headerMapper must not be null"); + this.headerMapper = headerMapper; + } + + protected HeaderMapper getHeaderMapper() { + return this.headerMapper; + } + + /** + * Specify the type of payload to be generated when the inbound HTTP request + * content is read by the converters/encoders. + * By default this value is null which means at runtime any "text" Content-Type will + * result in String while all others default to byte[].class. + * @param requestPayloadType The payload type. + */ + public void setRequestPayloadTypeClass(Class requestPayloadType) { + setRequestPayloadType(ResolvableType.forClass(requestPayloadType)); + } + + /** + * Specify the type of payload to be generated when the inbound HTTP request + * content is read by the converters/encoders. + * By default this value is null which means at runtime any "text" Content-Type will + * result in String while all others default to byte[].class. + * @param requestPayloadType The payload type. + */ + public void setRequestPayloadType(ResolvableType requestPayloadType) { + this.requestPayloadType = requestPayloadType; + } + + protected ResolvableType getRequestPayloadType() { + return this.requestPayloadType; + } + + /** + * Specify whether only the reply Message's payload should be passed in the response. + * If this is set to 'false', the entire Message will be used to generate the response. + * The default is 'true'. + * @param extractReplyPayload true to extract the reply payload. + */ + public void setExtractReplyPayload(boolean extractReplyPayload) { + this.extractReplyPayload = extractReplyPayload; + } + + protected boolean getExtractReplyPayload() { + return this.extractReplyPayload; + } + + /** + * Specify the {@link Expression} to resolve a status code for Response to override + * the default '200 OK' or '500 Internal Server Error' for a timeout. + * @param statusCodeExpression The status code Expression. + * @since 5.0 + * @see #setStatusCodeExpression(Expression) + */ + public void setStatusCodeExpressionString(String statusCodeExpression) { + setStatusCodeExpression(EXPRESSION_PARSER.parseExpression(statusCodeExpression)); + } + + /** + * Specify the {@link Expression} to resolve a status code for Response to override + * the default '200 OK' or '500 Internal Server Error' for a timeout. + *

The {@link #statusCodeExpression} is applied only for the one-way + * {@code } or when no reply (timeout) is received for + * a gateway. The {@code } (or whenever + * {@link #BaseHttpInboundEndpoint(boolean) expectReply} is true) resolves + * an {@link HttpStatus} from the + * {@link org.springframework.integration.http.HttpHeaders#STATUS_CODE} reply + * {@link Message} header. + * @param statusCodeExpression The status code Expression. + * @since 4.1 + * @see #setReplyTimeout(long) + * @see HttpRequestHandlingEndpointSupport#HttpRequestHandlingEndpointSupport(boolean) + */ + public void setStatusCodeExpression(Expression statusCodeExpression) { + this.statusCodeExpression = statusCodeExpression; + } + + protected Expression getStatusCodeExpression() { + return this.statusCodeExpression; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + + validateSupportedMethods(); + + if (this.statusCodeExpression != null) { + this.evaluationContext = createEvaluationContext(); + } + + getRequestMapping().setName(getComponentName()); + } + + private void validateSupportedMethods() { + if (this.requestPayloadType != null + && CollectionUtils.containsAny(nonReadableBodyHttpMethods, + Arrays.asList(getRequestMapping().getMethods()))) { + if (logger.isWarnEnabled()) { + logger.warn("The 'requestPayloadType' attribute will have no relevance for one " + + "of the specified HTTP methods '" + nonReadableBodyHttpMethods + "'"); + } + } + } + + protected HttpStatus evaluateHttpStatus() { + Object value = this.statusCodeExpression.getValue(this.evaluationContext); + return buildHttpStatus(value); + } + + protected HttpStatus resolveHttpStatusFromHeaders(MessageHeaders headers) { + Object httpStatusFromHeader = headers.get(org.springframework.integration.http.HttpHeaders.STATUS_CODE); + return buildHttpStatus(httpStatusFromHeader); + } + + private HttpStatus buildHttpStatus(Object httpStatusValue) { + HttpStatus httpStatus = null; + if (httpStatusValue instanceof HttpStatus) { + httpStatus = (HttpStatus) httpStatusValue; + } + else if (httpStatusValue instanceof Integer) { + httpStatus = HttpStatus.valueOf((Integer) httpStatusValue); + } + else if (httpStatusValue instanceof String) { + httpStatus = HttpStatus.valueOf(Integer.parseInt((String) httpStatusValue)); + } + return httpStatus; + } + + protected StandardEvaluationContext createEvaluationContext() { + return ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + } + + @Override + public int beforeShutdown() { + stop(); + return this.activeCount.get(); + } + + @Override + public int afterShutdown() { + return this.activeCount.get(); + } + + @Override + public String getComponentType() { + return (this.expectReply) ? "http:inbound-gateway" : "http:inbound-channel-adapter"; + } + + /** + * Checks if the request has a readable body (not a GET, HEAD, or OPTIONS request). + * @param request the HTTP request to check the method + * @return true or false if HTTP request can contain the body + */ + protected boolean isReadable(HttpRequest request) { + return !(CollectionUtils.containsInstance(nonReadableBodyHttpMethods, request.getMethod())); + } +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java index cdfbdbe0b2..2ac91d66c3 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java @@ -18,12 +18,10 @@ package org.springframework.integration.http.inbound; import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; -import java.util.concurrent.atomic.AtomicInteger; import javax.servlet.http.Cookie; import javax.servlet.http.HttpServletRequest; @@ -32,12 +30,10 @@ import javax.xml.transform.Source; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.NoSuchBeanDefinitionException; -import org.springframework.expression.EvaluationContext; +import org.springframework.core.ResolvableType; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.http.HttpEntity; -import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.converter.ByteArrayHttpMessageConverter; @@ -49,23 +45,18 @@ import org.springframework.http.converter.feed.RssChannelHttpMessageConverter; import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; import org.springframework.http.converter.xml.Jaxb2RootElementHttpMessageConverter; import org.springframework.http.converter.xml.SourceHttpMessageConverter; +import org.springframework.http.server.ServerHttpResponse; import org.springframework.http.server.ServletServerHttpRequest; import org.springframework.http.server.ServletServerHttpResponse; import org.springframework.integration.MessageTimeoutException; -import org.springframework.integration.context.OrderlyShutdownCapable; -import org.springframework.integration.expression.ExpressionUtils; -import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.http.converter.MultipartAwareFormHttpMessageConverter; import org.springframework.integration.http.multipart.MultipartHttpInputMessage; -import org.springframework.integration.http.support.DefaultHttpHeaderMapper; import org.springframework.integration.mapping.HeaderMapper; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.json.JacksonPresent; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; @@ -108,52 +99,21 @@ import org.springframework.web.servlet.HandlerMapping; * @author Gary Russell * @author Artem Bilan * @author Biju Kunjummen + * * @since 2.0 */ -public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySupport - implements OrderlyShutdownCapable { +public abstract class HttpRequestHandlingEndpointSupport extends BaseHttpInboundEndpoint { - private static final boolean jaxb2Present = ClassUtils.isPresent("javax.xml.bind.Binder", - HttpRequestHandlingEndpointSupport.class.getClassLoader()); + private final List> defaultMessageConverters = new ArrayList<>(); - private static final boolean romeToolsPresent = ClassUtils.isPresent("com.rometools.rome.feed.atom.Feed", - HttpRequestHandlingEndpointSupport.class.getClassLoader()); - - private static final List nonReadableBodyHttpMethods = - Arrays.asList(HttpMethod.GET, HttpMethod.HEAD, HttpMethod.OPTIONS); - - private final List> defaultMessageConverters = new ArrayList>(); - - private final boolean expectReply; - - private volatile List> messageConverters = new ArrayList>(); - - private volatile RequestMapping requestMapping = new RequestMapping(); - - private volatile CrossOrigin crossOrigin; - - private volatile Class requestPayloadType = null; + private volatile List> messageConverters = new ArrayList<>(); private volatile boolean convertersMerged; private volatile boolean mergeWithDefaultConverters = false; - private volatile HeaderMapper headerMapper = DefaultHttpHeaderMapper.inboundMapper(); - - private volatile boolean extractReplyPayload = true; - private volatile MultipartResolver multipartResolver; - private volatile Expression payloadExpression; - - private volatile Map headerExpressions; - - private volatile Expression statusCodeExpression; - - private volatile EvaluationContext evaluationContext; - - private final AtomicInteger activeCount = new AtomicInteger(); - /** * Construct a gateway that will wait for the {@link #setReplyTimeout(long) * replyTimeout} for a reply; if the timeout is exceeded a '500 Internal Server Error' @@ -179,7 +139,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa */ public HttpRequestHandlingEndpointSupport(boolean expectReply) { super(expectReply); - this.expectReply = expectReply; this.defaultMessageConverters.add(new MultipartAwareFormHttpMessageConverter()); this.defaultMessageConverters.add(new ByteArrayHttpMessageConverter()); StringHttpMessageConverter stringHttpMessageConverter = new StringHttpMessageConverter(); @@ -210,50 +169,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa } } - /** - * @return Whether to expect a reply. - */ - protected boolean isExpectReply() { - return this.expectReply; - } - - /** - * Specifies a SpEL expression to evaluate in order to generate the Message payload. - * The EvaluationContext will be populated with an HttpEntity instance as the root object, - * and it may contain variables: - *

    - *
  • #pathVariables
    • - *
  • #requestParams
    • - *
  • #requestAttributes
    • - *
  • #requestHeaders
    • - *
  • #matrixVariables
    • - *
  • #cookies - *
- * @param payloadExpression The payload expression. - */ - public void setPayloadExpression(Expression payloadExpression) { - this.payloadExpression = payloadExpression; - } - - /** - * Specifies a Map of SpEL expressions to evaluate in order to generate the Message headers. - * The keys in the map will be used as the header names. When evaluating the expression, - * the EvaluationContext will be populated with an HttpEntity instance as the root object, - * and it may contain variables: - *
    - *
  • #pathVariables
    • - *
  • #requestParams
    • - *
  • #requestAttributes
    • - *
  • #requestHeaders
    • - *
  • #matrixVariables
    • - *
  • #cookies - *
- * @param headerExpressions The header expressions. - */ - public void setHeaderExpressions(Map headerExpressions) { - this.headerExpressions = headerExpressions; - } - /** * Set the message body converters to use. These converters are used to convert from and to HTTP requests and * responses. @@ -283,60 +198,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa this.mergeWithDefaultConverters = mergeWithDefaultConverters; } - /** - * Set the {@link HeaderMapper} to use when mapping between HTTP headers and MessageHeaders. - * @param headerMapper The header mapper. - */ - public void setHeaderMapper(HeaderMapper headerMapper) { - Assert.notNull(headerMapper, "headerMapper must not be null"); - this.headerMapper = headerMapper; - } - - /** - * Set the {@link RequestMapping} which allows you to specify a flexible RESTFul-mapping for this endpoint. - * @param requestMapping The request mapping. - */ - public void setRequestMapping(RequestMapping requestMapping) { - Assert.notNull(requestMapping, "requestMapping must not be null"); - this.requestMapping = requestMapping; - } - - public final RequestMapping getRequestMapping() { - return this.requestMapping; - } - - - /** - * Set the {@link CrossOrigin} to permit cross origin requests for this endpoint. - * @param crossOrigin the CrossOrigin config. - * @since 4.2 - */ - public void setCrossOrigin(CrossOrigin crossOrigin) { - this.crossOrigin = crossOrigin; - } - - public CrossOrigin getCrossOrigin() { - return this.crossOrigin; - } - - /** - * Specify the type of payload to be generated when the inbound HTTP request content is read by the - * {@link HttpMessageConverter}s. By default this value is null which means at runtime any "text" Content-Type will - * result in String while all others default to byte[].class. - * @param requestPayloadType The payload type. - */ - public void setRequestPayloadType(Class requestPayloadType) { - this.requestPayloadType = requestPayloadType; - } - - /** - * Specify whether only the reply Message's payload should be passed in the response. If this is set to 'false', the - * entire Message will be used to generate the response. The default is 'true'. - * @param extractReplyPayload true to extract the reply payload. - */ - public void setExtractReplyPayload(boolean extractReplyPayload) { - this.extractReplyPayload = extractReplyPayload; - } /** * Specify the {@link MultipartResolver} to use when checking requests. If no resolver is provided, the @@ -347,40 +208,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa public void setMultipartResolver(MultipartResolver multipartResolver) { this.multipartResolver = multipartResolver; } - /** - * Specify the {@link Expression} to resolve a status code for Response to override - * the default '200 OK' or '500 Internal Server Error' for a timeout. - * @param statusCodeExpression The status code Expression. - * @since 5.0 - * @see #setStatusCodeExpression(Expression) - */ - public void setStatusCodeExpressionString(String statusCodeExpression) { - setStatusCodeExpression(EXPRESSION_PARSER.parseExpression(statusCodeExpression)); - } - - /** - * Specify the {@link Expression} to resolve a status code for Response to override - * the default '200 OK' or '500 Internal Server Error' for a timeout. - *

The {@link #statusCodeExpression} is applied only for the one-way - * {@code } or when no reply (timeout) is received for - * a gateway. The {@code } (or whenever - * {@link #HttpRequestHandlingEndpointSupport(boolean) expectReply} is true) resolves - * an {@link HttpStatus} from the - * {@link org.springframework.integration.http.HttpHeaders#STATUS_CODE} reply - * {@link Message} header. - * @param statusCodeExpression The status code Expression. - * @since 4.1 - * @see #setReplyTimeout(long) - * @see HttpRequestHandlingEndpointSupport#HttpRequestHandlingEndpointSupport(boolean) - */ - public void setStatusCodeExpression(Expression statusCodeExpression) { - this.statusCodeExpression = statusCodeExpression; - } - - @Override - public String getComponentType() { - return (this.expectReply) ? "http:inbound-gateway" : "http:inbound-channel-adapter"; - } /** * Locates the {@link MultipartResolver} bean based on the default name defined by the @@ -391,7 +218,7 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa @Override protected void onInit() throws Exception { super.onInit(); - BeanFactory beanFactory = this.getBeanFactory(); + BeanFactory beanFactory = getBeanFactory(); if (this.multipartResolver == null && beanFactory != null) { try { MultipartResolver multipartResolver = beanFactory.getBean( @@ -412,11 +239,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa if (this.messageConverters.size() == 0 || (this.mergeWithDefaultConverters && !this.convertersMerged)) { this.messageConverters.addAll(this.defaultMessageConverters); } - this.validateSupportedMethods(); - - if (this.statusCodeExpression != null) { - this.evaluationContext = createEvaluationContext(); - } } /** @@ -437,7 +259,7 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa } } - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings({ "rawtypes", "unchecked" }) private Message actualDoHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws IOException { this.activeCount.incrementAndGet(); @@ -445,8 +267,8 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa ServletServerHttpRequest request = this.prepareRequest(servletRequest); Object requestBody = null; - if (this.isReadable(request)) { - requestBody = this.extractRequestBody(request); + if (isReadable(request)) { + requestBody = extractRequestBody(request); } HttpEntity httpEntity = new HttpEntity(requestBody, request.getHeaders()); @@ -490,14 +312,14 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa evaluationContext.setVariable("matrixVariables", matrixVariables); } - Map headers = this.headerMapper.toHeaders(request.getHeaders()); + Map headers = getHeaderMapper().toHeaders(request.getHeaders()); Object payload = null; - if (this.payloadExpression != null) { + if (getPayloadExpression() != null) { // create payload based on SpEL - payload = this.payloadExpression.getValue(evaluationContext); + payload = getPayloadExpression().getValue(evaluationContext); } - if (!CollectionUtils.isEmpty(this.headerExpressions)) { - for (Entry entry : this.headerExpressions.entrySet()) { + if (!CollectionUtils.isEmpty(getHeaderExpressions())) { + for (Entry entry : getHeaderExpressions().entrySet()) { String headerName = entry.getKey(); Expression headerExpression = entry.getValue(); Object headerValue = headerExpression.getValue(evaluationContext); @@ -541,11 +363,11 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa reply = this.sendAndReceiveMessage(message); } catch (MessageTimeoutException e) { - if (this.statusCodeExpression != null) { + if (getStatusCodeExpression() != null) { reply = getMessageBuilderFactory().withPayload(e.getMessage()) - .setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, - evaluateHttpStatus()) - .build(); + .setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, + evaluateHttpStatus()) + .build(); } else { reply = getMessageBuilderFactory().withPayload(e.getMessage()) @@ -583,23 +405,22 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa * @return The message payload (if {@link #extractReplyPayload}) otherwise the message. */ protected final Object setupResponseAndConvertReply(ServletServerHttpResponse response, Message replyMessage) { - - this.headerMapper.fromHeaders(replyMessage.getHeaders(), response.getHeaders()); + getHeaderMapper().fromHeaders(replyMessage.getHeaders(), response.getHeaders()); HttpStatus httpStatus = this.resolveHttpStatusFromHeaders(replyMessage.getHeaders()); if (httpStatus != null) { response.setStatusCode(httpStatus); } Object reply = replyMessage; - if (this.extractReplyPayload) { + if (getExtractReplyPayload()) { reply = replyMessage.getPayload(); } return reply; } - protected void setStatusCodeIfNeeded(ServletServerHttpResponse response) { - if (this.statusCodeExpression != null) { + protected void setStatusCodeIfNeeded(ServerHttpResponse response) { + if (getStatusCodeExpression() != null) { HttpStatus httpStatus = evaluateHttpStatus(); if (httpStatus != null) { response.setStatusCode(httpStatus); @@ -607,14 +428,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa } } - private HttpStatus evaluateHttpStatus() { - if (this.evaluationContext == null) { - this.evaluationContext = createEvaluationContext(); - } - Object value = this.statusCodeExpression.getValue(this.evaluationContext); - return buildHttpStatus(value); - } - /** * Prepares an instance of {@link ServletServerHttpRequest} from the raw * {@link HttpServletRequest}. Also converts the request into a multipart request to @@ -634,13 +447,6 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa return new ServletServerHttpRequest(servletRequest); } - /** - * Checks if the request has a readable body (not a GET, HEAD, or OPTIONS request). - */ - private boolean isReadable(ServletServerHttpRequest request) { - return !(CollectionUtils.containsInstance(nonReadableBodyHttpMethods, request.getMethod())); - } - /** * Clean up any resources used by the given multipart request (if any). * @param request current HTTP request @@ -666,16 +472,21 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa return convertedMap; } - @SuppressWarnings({"unchecked", "rawtypes"}) + @SuppressWarnings({ "unchecked", "rawtypes" }) private Object extractRequestBody(ServletServerHttpRequest request) throws IOException { MediaType contentType = request.getHeaders().getContentType(); if (contentType == null) { contentType = MediaType.APPLICATION_OCTET_STREAM; } - Class expectedType = this.requestPayloadType; - if (expectedType == null) { - expectedType = ("text".equals(contentType.getType())) ? String.class : byte[].class; + ResolvableType requestPayloadType = getRequestPayloadType(); + Class expectedType; + if (requestPayloadType == null) { + expectedType = "text".equals(contentType.getType()) ? String.class : byte[].class; } + else { + expectedType = requestPayloadType.resolve(); + } + for (HttpMessageConverter converter : this.messageConverters) { if (converter.canRead(expectedType, contentType)) { return converter.read((Class) expectedType, request); @@ -686,50 +497,4 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa + expectedType.getName() + "] and content type [" + contentType + "]"); } - private HttpStatus resolveHttpStatusFromHeaders(MessageHeaders headers) { - Object httpStatusFromHeader = headers.get(org.springframework.integration.http.HttpHeaders.STATUS_CODE); - return buildHttpStatus(httpStatusFromHeader); - } - - private HttpStatus buildHttpStatus(Object httpStatusValue) { - HttpStatus httpStatus = null; - if (httpStatusValue instanceof HttpStatus) { - httpStatus = (HttpStatus) httpStatusValue; - } - else if (httpStatusValue instanceof Integer) { - httpStatus = HttpStatus.valueOf((Integer) httpStatusValue); - } - else if (httpStatusValue instanceof String) { - httpStatus = HttpStatus.valueOf(Integer.parseInt((String) httpStatusValue)); - } - return httpStatus; - } - - protected StandardEvaluationContext createEvaluationContext() { - return ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory()); - } - - private void validateSupportedMethods() { - if (this.requestPayloadType != null - && CollectionUtils.containsAny(nonReadableBodyHttpMethods, - Arrays.asList(this.requestMapping.getMethods()))) { - if (logger.isWarnEnabled()) { - logger.warn("The 'requestPayloadType' attribute will have no relevance for one " + - "of the specified HTTP methods '" + nonReadableBodyHttpMethods + "'"); - } - } - } - - - @Override - public int beforeShutdown() { - stop(); - return this.activeCount.get(); - } - - @Override - public int afterShutdown() { - return this.activeCount.get(); - } - } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGateway.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGateway.java index 7564e02bfe..f470c99c58 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGateway.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGateway.java @@ -46,7 +46,7 @@ import org.springframework.web.HttpRequestHandler; * The default supported request methods are GET and POST, but the list of values can be configured with the * {@link RequestMapping#methods} property. The payload generated from a GET request (or HEAD or OPTIONS if supported) will * be a {@link MultiValueMap} containing the parameter values. For a request containing a body (e.g. a POST), the type - * of the payload is determined by the {@link #setRequestPayloadType(Class) request payload type}. + * of the payload is determined by the {@link #setRequestPayloadTypeClass(Class)} request payload type}. *

* If the HTTP request is a multipart and a "multipartResolver" bean has been defined in the context, then it will be * converted by the {@link MultipartAwareFormHttpMessageConverter} as long as the default message converters have not diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java new file mode 100644 index 0000000000..5488cfb696 --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationHandlerResultHandler.java @@ -0,0 +1,58 @@ +/* + * 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.inbound; + +import org.springframework.core.Ordered; +import org.springframework.web.method.HandlerMethod; +import org.springframework.web.reactive.HandlerResult; +import org.springframework.web.reactive.HandlerResultHandler; +import org.springframework.web.server.ServerWebExchange; + +import reactor.core.publisher.Mono; + +/** + * A {@link HandlerResultHandler} implementation to handle the result of the + * {@link ReactiveHttpInboundEndpoint} execution. Actually just return the + * {@code result.getReturnValue()} which essentially is expected {@code Mono}. + * + * @author Artem Bilan + * + * @since 5.0 + * + * @see ReactiveHttpInboundEndpoint + */ +public class IntegrationHandlerResultHandler implements HandlerResultHandler, Ordered { + + @Override + public boolean supports(HandlerResult result) { + Object handler = result.getHandler(); + return handler instanceof HandlerMethod + && ReactiveHttpInboundEndpoint.class.isAssignableFrom(((HandlerMethod) handler).getBeanType()); + } + + @Override + @SuppressWarnings("unchecked") + public Mono handleResult(ServerWebExchange exchange, HandlerResult result) { + return (Mono) result.getReturnValue(); + } + + @Override + public int getOrder() { + return HIGHEST_PRECEDENCE; + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java index 71f9c30da5..6d99fb718c 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-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. @@ -50,7 +50,7 @@ import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandl * elements, if there is none registered, yet. However it can be configured as a regular * bean with appropriate configuration for {@link RequestMappingHandlerMapping}. * It is recommended to have only one similar bean in the application context using the 'id' - * {@link org.springframework.integration.http.support.HttpContextUtils#HANDLER_MAPPING_BEAN_NAME}. + * {@link org.springframework.integration.http.support.HttpContextUtils#REACTIVE_HANDLER_MAPPING_BEAN_NAME}. *

* In most cases, Spring MVC offers to configure Request Mapping via * {@code org.springframework.stereotype.Controller} and @@ -69,7 +69,9 @@ import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandl * (e.g. Spring Integration XML) configurations. * * @author Artem Bilan + * * @since 3.0 + * * @see RequestMapping * @see RequestMappingHandlerMapping */ @@ -113,7 +115,7 @@ public final class IntegrationRequestMappingHandlerMapping extends RequestMappin if (handler instanceof String) { handler = this.getApplicationContext().getBean((String) handler); } - RequestMappingInfo mapping = this.getMappingForEndpoint((HttpRequestHandlingEndpointSupport) handler); + RequestMappingInfo mapping = this.getMappingForEndpoint((BaseHttpInboundEndpoint) handler); if (mapping != null) { registerMapping(mapping, handler, HANDLE_REQUEST_METHOD); } @@ -121,7 +123,7 @@ public final class IntegrationRequestMappingHandlerMapping extends RequestMappin @Override protected CorsConfiguration initCorsConfiguration(Object handler, Method method, RequestMappingInfo mappingInfo) { - CrossOrigin crossOrigin = ((HttpRequestHandlingEndpointSupport) handler).getCrossOrigin(); + CrossOrigin crossOrigin = ((BaseHttpInboundEndpoint) handler).getCrossOrigin(); if (crossOrigin != null) { CorsConfiguration config = new CorsConfiguration(); for (String origin : crossOrigin.getOrigin()) { @@ -154,7 +156,7 @@ public final class IntegrationRequestMappingHandlerMapping extends RequestMappin } } } - return config; + return config.applyPermitDefaultValues(); } return null; } @@ -164,7 +166,7 @@ public final class IntegrationRequestMappingHandlerMapping extends RequestMappin * 'Spring Integration HTTP Inbound Endpoint' {@link RequestMapping}. * @see RequestMappingHandlerMapping#getMappingForMethod */ - private RequestMappingInfo getMappingForEndpoint(HttpRequestHandlingEndpointSupport endpoint) { + private RequestMappingInfo getMappingForEndpoint(BaseHttpInboundEndpoint endpoint) { final RequestMapping requestMapping = endpoint.getRequestMapping(); if (ObjectUtils.isEmpty(requestMapping.getPathPatterns())) { diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java new file mode 100644 index 0000000000..df25d49293 --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpoint.java @@ -0,0 +1,461 @@ +/* + * 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.inbound; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import org.reactivestreams.Publisher; + +import org.springframework.core.ReactiveAdapter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.ResolvableType; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.codec.HttpMessageReader; +import org.springframework.http.codec.HttpMessageWriter; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.http.server.reactive.ServerHttpRequest; +import org.springframework.http.server.reactive.ServerHttpResponse; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.MultiValueMap; +import org.springframework.web.reactive.HandlerMapping; +import org.springframework.web.reactive.accept.HeaderContentTypeResolver; +import org.springframework.web.reactive.accept.RequestedContentTypeResolver; +import org.springframework.web.server.NotAcceptableStatusException; +import org.springframework.web.server.ServerWebExchange; +import org.springframework.web.server.UnsupportedMediaTypeStatusException; +import org.springframework.web.server.WebHandler; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * A {@link MessagingGatewaySupport} implementation for Spring WebFlux + * HTTP requests execution. + * + * @author Artem Bilan + * + * @since 5.0 + * + * @see org.springframework.web.reactive.result.HandlerResultHandlerSupport + * @see org.springframework.web.reactive.config.EnableWebFlux + */ +public class ReactiveHttpInboundEndpoint extends BaseHttpInboundEndpoint implements WebHandler { + + private static final MediaType MEDIA_TYPE_APPLICATION_ALL = new MediaType("application"); + + private ServerCodecConfigurer codecConfigurer = ServerCodecConfigurer.create(); + + private RequestedContentTypeResolver requestedContentTypeResolver = new HeaderContentTypeResolver(); + + private ReactiveAdapterRegistry adapterRegistry = new ReactiveAdapterRegistry(); + + public ReactiveHttpInboundEndpoint() { + this(true); + } + + public ReactiveHttpInboundEndpoint(boolean expectReply) { + super(expectReply); + } + + /** + * A {@link ServerCodecConfigurer} for the request readers and response writers. + * By default the {@link ServerCodecConfigurer#create()} factory is used. + * @param codecConfigurer the {@link ServerCodecConfigurer} to use. + */ + public void setCodecConfigurer(ServerCodecConfigurer codecConfigurer) { + Assert.notNull(codecConfigurer, "'codecConfigurer' must not be null"); + this.codecConfigurer = codecConfigurer; + } + + /** + * A strategy to resolve the requested media types for a {@code ServerWebExchange}. + * A {@link HeaderContentTypeResolver} is used by default. + * @param requestedContentTypeResolver the {@link RequestedContentTypeResolver} to use. + */ + public void setRequestedContentTypeResolver(RequestedContentTypeResolver requestedContentTypeResolver) { + Assert.notNull(requestedContentTypeResolver, "'requestedContentTypeResolver' must not be null"); + this.requestedContentTypeResolver = requestedContentTypeResolver; + } + + /** + * A registry of adapters to adapt a Reactive Streams {@link Publisher} to/from. + * @param adapterRegistry the {@link ReactiveAdapterRegistry} to use. + */ + public void setReactiveAdapterRegistry(ReactiveAdapterRegistry adapterRegistry) { + Assert.notNull(adapterRegistry, "'adapterRegistry' must not be null"); + this.adapterRegistry = adapterRegistry; + } + + @Override + public String getComponentType() { + return super.getComponentType().replaceFirst("(http:)", "$1reactive-"); + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + } + + @Override + public Mono handle(ServerWebExchange exchange) { + return Mono.defer(() -> { + if (isRunning()) { + return doHandle(exchange); + } + else { + return serviceUnavailableResponse(exchange); + } + }); + } + + @SuppressWarnings("unchecked") + private Mono doHandle(ServerWebExchange exchange) { + return extractRequestBody(exchange) + .doOnSubscribe(s -> this.activeCount.incrementAndGet()) + .map(body -> new HttpEntity<>(body, exchange.getRequest().getHeaders())) + .map(entity -> buildMessage(entity, exchange)) + .flatMap(requestMessage -> { + if (this.expectReply) { + return sendAndReceiveMessageReactive(requestMessage) + .flatMap(replyMessage -> populateResponse(exchange, replyMessage)); + } + else { + send(requestMessage); + return setStatusCode(exchange); + } + }) + .doOnTerminate((e, t) -> this.activeCount.decrementAndGet()); + + } + + @SuppressWarnings("unchecked") + private Mono extractRequestBody(ServerWebExchange exchange) { + ServerHttpRequest request = exchange.getRequest(); + ServerHttpResponse response = exchange.getResponse(); + + if (isReadable(request)) { + MediaType contentType; + if (request.getHeaders().getContentType() == null) { + contentType = MediaType.APPLICATION_OCTET_STREAM; + } + else { + contentType = request.getHeaders().getContentType(); + } + + if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(contentType)) { + return (Mono) exchange.getFormData(); + } + else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) { + return (Mono) exchange.getMultipartData(); + } + else { + ResolvableType bodyType = getRequestPayloadType(); + if (bodyType == null) { + bodyType = + "text".equals(contentType.getType()) + ? ResolvableType.forClass(String.class) + : ResolvableType.forClass(byte[].class); + } + + Class resolvedType = bodyType.resolve(); + + ReactiveAdapter adapter = (resolvedType != null ? this.adapterRegistry.getAdapter(resolvedType) : null); + ResolvableType elementType = (adapter != null ? bodyType.getGeneric() : bodyType); + + HttpMessageReader httpMessageReader = this.codecConfigurer + .getReaders() + .stream() + .filter(reader -> reader.canRead(elementType, contentType)) + .findFirst() + .orElseThrow(() -> new UnsupportedMediaTypeStatusException( + "Could not convert request: no suitable HttpMessageReader found for expected type [" + + elementType + "] and content type [" + contentType + "]")); + + + Map readHints = Collections.emptyMap(); + if (adapter != null && adapter.isMultiValue()) { + Flux flux = httpMessageReader.read(bodyType, elementType, request, response, readHints); + + return (Mono) Mono.just(adapter.fromPublisher(flux)); + } + else { + Mono mono = httpMessageReader.readMono(bodyType, elementType, request, response, readHints); + + if (adapter != null) { + return (Mono) Mono.just(adapter.fromPublisher(mono)); + } + else { + return (Mono) mono; + } + } + } + } + else { + return (Mono) Mono.just(exchange.getRequest().getQueryParams()); + } + } + + @SuppressWarnings("unchecked") + private Message buildMessage(HttpEntity httpEntity, ServerWebExchange exchange) { + ServerHttpRequest request = exchange.getRequest(); + HttpHeaders requestHeaders = request.getHeaders(); + Map exchangeAttributes = exchange.getAttributes(); + + StandardEvaluationContext evaluationContext = createEvaluationContext(); + + evaluationContext.setVariable("requestAttributes", exchangeAttributes); + MultiValueMap requestParams = request.getQueryParams(); + evaluationContext.setVariable("requestParams", requestParams); + evaluationContext.setVariable("requestHeaders", requestHeaders); + if (!CollectionUtils.isEmpty(request.getCookies())) { + evaluationContext.setVariable("cookies", request.getCookies()); + } + + Map pathVariables = + (Map) exchangeAttributes.get(HandlerMapping.URI_TEMPLATE_VARIABLES_ATTRIBUTE); + + if (!CollectionUtils.isEmpty(pathVariables)) { + evaluationContext.setVariable("pathVariables", pathVariables); + } + + Map> matrixVariables = + (Map>) exchangeAttributes.get(HandlerMapping.MATRIX_VARIABLES_ATTRIBUTE); + + if (!CollectionUtils.isEmpty(matrixVariables)) { + evaluationContext.setVariable("matrixVariables", matrixVariables); + } + + evaluationContext.setRootObject(httpEntity); + Object payload; + if (getPayloadExpression() != null) { + payload = getPayloadExpression().getValue(evaluationContext); + if (payload == null) { + throw new IllegalStateException("The payload expression '" + getPayloadExpression().getExpressionString() + + "' returned null."); + } + } + else { + payload = httpEntity.getBody(); + } + + Map headers = getHeaderMapper().toHeaders(request.getHeaders()); + if (!CollectionUtils.isEmpty(getHeaderExpressions())) { + for (Map.Entry entry : getHeaderExpressions().entrySet()) { + String headerName = entry.getKey(); + Expression headerExpression = entry.getValue(); + Object headerValue = headerExpression.getValue(evaluationContext); + if (headerValue != null) { + headers.put(headerName, headerValue); + } + } + } + + AbstractIntegrationMessageBuilder messageBuilder; + + if (payload instanceof Message) { + messageBuilder = + getMessageBuilderFactory() + .fromMessage((Message) payload) + .copyHeadersIfAbsent(headers); + } + else { + messageBuilder = + getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headers); + } + + return messageBuilder + .setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL, request.getURI().toString()) + .setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD, request.getMethod().toString()) + .setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL, exchange.getPrincipal().block()) + .build(); + } + + private Mono populateResponse(ServerWebExchange exchange, Message replyMessage) { + ServerHttpResponse response = exchange.getResponse(); + getHeaderMapper().fromHeaders(replyMessage.getHeaders(), response.getHeaders()); + + Object responseContent = replyMessage; + if (getExtractReplyPayload()) { + responseContent = replyMessage.getPayload(); + } + + if (responseContent instanceof HttpStatus) { + response.setStatusCode((HttpStatus) responseContent); + return response.setComplete(); + } + else { + HttpStatus httpStatus = resolveHttpStatusFromHeaders(replyMessage.getHeaders()); + if (httpStatus != null) { + response.setStatusCode(httpStatus); + } + + return writeResponseBody(exchange, responseContent); + } + } + + @SuppressWarnings("unchecked") + private Mono writeResponseBody(ServerWebExchange exchange, Object body) { + ResolvableType bodyType = ResolvableType.forInstance(body); + ReactiveAdapter adapter = this.adapterRegistry.getAdapter(bodyType.resolve(), body); + + Publisher publisher; + ResolvableType elementType; + if (adapter != null) { + publisher = adapter.toPublisher(body); + ResolvableType genericType = bodyType.getGeneric(0); + elementType = getElementType(adapter, genericType); + } + else { + publisher = Mono.justOrEmpty(body); + elementType = bodyType; + } + + if (void.class == elementType.getRawClass() || Void.class == elementType.getRawClass()) { + return Mono.from((Publisher) publisher); + } + + List producibleMediaTypes = getProducibleMediaTypes(bodyType); + MediaType bestMediaType = selectMediaType(exchange, () -> producibleMediaTypes); + + if (bestMediaType != null) { + for (HttpMessageWriter writer : this.codecConfigurer.getWriters()) { + if (writer.canWrite(bodyType, bestMediaType)) { + return ((HttpMessageWriter) writer).write(publisher, elementType, + bestMediaType, exchange.getResponse(), Collections.emptyMap()); + } + } + } + else { + if (producibleMediaTypes.isEmpty()) { + return Mono.error(new IllegalStateException("No HttpMessageWriters for response type: " + bodyType)); + } + } + + return Mono.error(new NotAcceptableStatusException(producibleMediaTypes)); + } + + private ResolvableType getElementType(ReactiveAdapter adapter, ResolvableType genericType) { + if (adapter.isNoValue()) { + return ResolvableType.forClass(Void.class); + } + else if (genericType != ResolvableType.NONE) { + return genericType; + } + else { + return ResolvableType.forClass(Object.class); + } + } + + private List getProducibleMediaTypes(ResolvableType elementType) { + return this.codecConfigurer.getWriters() + .stream() + .filter(converter -> converter.canWrite(elementType, null)) + .flatMap(converter -> converter.getWritableMediaTypes().stream()) + .collect(Collectors.toList()); + } + + private MediaType selectMediaType(ServerWebExchange exchange, Supplier> producibleTypesSupplier) { + List acceptableTypes = getAcceptableTypes(exchange); + List producibleTypes = getProducibleTypes(exchange, producibleTypesSupplier); + + Set compatibleMediaTypes = new LinkedHashSet<>(); + for (MediaType acceptable : acceptableTypes) { + for (MediaType producible : producibleTypes) { + if (acceptable.isCompatibleWith(producible)) { + compatibleMediaTypes.add(selectMoreSpecificMediaType(acceptable, producible)); + } + } + } + + List result = new ArrayList<>(compatibleMediaTypes); + MediaType.sortBySpecificityAndQuality(result); + + for (MediaType mediaType : result) { + if (mediaType.isConcrete()) { + return mediaType; + } + else if (mediaType.equals(MediaType.ALL) || mediaType.equals(MEDIA_TYPE_APPLICATION_ALL)) { + return MediaType.APPLICATION_OCTET_STREAM; + } + } + + return null; + } + + private List getAcceptableTypes(ServerWebExchange exchange) { + List mediaTypes = this.requestedContentTypeResolver.resolveMediaTypes(exchange); + return (mediaTypes.isEmpty() ? Collections.singletonList(MediaType.ALL) : mediaTypes); + } + + @SuppressWarnings("unchecked") + private List getProducibleTypes(ServerWebExchange exchange, + Supplier> producibleTypesSupplier) { + + Set mediaTypes = exchange.getAttribute(HandlerMapping.PRODUCIBLE_MEDIA_TYPES_ATTRIBUTE); + return (mediaTypes != null ? new ArrayList<>(mediaTypes) : producibleTypesSupplier.get()); + } + + private MediaType selectMoreSpecificMediaType(MediaType acceptable, MediaType producible) { + producible = producible.copyQualityValue(acceptable); + Comparator comparator = MediaType.SPECIFICITY_COMPARATOR; + return (comparator.compare(acceptable, producible) <= 0 ? acceptable : producible); + } + + + private Mono setStatusCode(ServerWebExchange exchange) { + ServerHttpResponse response = exchange.getResponse(); + if (getStatusCodeExpression() != null) { + HttpStatus httpStatus = evaluateHttpStatus(); + if (httpStatus != null) { + response.setStatusCode(httpStatus); + } + } + + return response.setComplete(); + } + + private Mono serviceUnavailableResponse(ServerWebExchange exchange) { + if (logger.isDebugEnabled()) { + logger.debug("Endpoint is stopped; returning status " + HttpStatus.SERVICE_UNAVAILABLE); + } + ServerHttpResponse response = exchange.getResponse(); + response.setStatusCode(HttpStatus.SERVICE_UNAVAILABLE); + return response.writeWith( + Mono.just(response.bufferFactory() + .wrap("Endpoint is stopped".getBytes()))); + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java new file mode 100644 index 0000000000..5edc84883d --- /dev/null +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/ReactiveIntegrationRequestMappingHandlerMapping.java @@ -0,0 +1,170 @@ +/* + * 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.inbound; + +import java.lang.reflect.Method; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.springframework.context.ApplicationListener; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.integration.http.support.HttpContextUtils; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.cors.CorsConfiguration; +import org.springframework.web.reactive.result.condition.NameValueExpression; +import org.springframework.web.reactive.result.method.RequestMappingInfo; +import org.springframework.web.reactive.result.method.annotation.RequestMappingHandlerMapping; +import org.springframework.web.server.ServerWebExchange; +import org.springframework.web.server.WebHandler; + +/** + * The {@link org.springframework.web.servlet.HandlerMapping} implementation that + * detects and registers {@link org.springframework.web.servlet.mvc.method.RequestMappingInfo}s for + * {@link HttpRequestHandlingEndpointSupport} from a Spring Integration HTTP configuration + * of {@code } and {@code } elements. + *

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

+ * In most cases, Spring MVC offers to configure Request Mapping via + * {@code org.springframework.stereotype.Controller} and + * {@link org.springframework.web.bind.annotation.RequestMapping}. + * That's why Spring MVC's Handler Mapping infrastructure relies on + * {@link org.springframework.web.method.HandlerMethod}, as different methods at the same + * {@code org.springframework.stereotype.Controller} user-class may have their own + * {@link org.springframework.web.bind.annotation.RequestMapping}. + * On the other side, all Spring Integration HTTP Inbound Endpoints are configured on + * the basis of the same {@link HttpRequestHandlingEndpointSupport} class and there is no + * single {@link org.springframework.web.reactive.result.method.RequestMappingInfo} configuration without + * {@link org.springframework.web.method.HandlerMethod} in Spring MVC. + * Accordingly {@link ReactiveIntegrationRequestMappingHandlerMapping} is a + * {@link org.springframework.web.reactive.HandlerMapping} + * compromise implementation between method-level annotations and component-level + * (e.g. Spring Integration XML) configurations. + * + * @author Artem Bilan + * + * @since 5.0 + * + * @see RequestMapping + * @see RequestMappingHandlerMapping + */ +public class ReactiveIntegrationRequestMappingHandlerMapping extends RequestMappingHandlerMapping + implements ApplicationListener { + + private static final Method HANDLER_METHOD = ReflectionUtils.findMethod(WebHandler.class, + "handle", ServerWebExchange.class); + + private final AtomicBoolean initialized = new AtomicBoolean(); + + @Override + protected boolean isHandler(Class beanType) { + return ReactiveHttpInboundEndpoint.class.isAssignableFrom(beanType); + } + + @Override + protected void detectHandlerMethods(Object handler) { + if (handler instanceof String) { + handler = getApplicationContext().getBean((String) handler); + } + RequestMappingInfo mapping = getMappingForEndpoint((ReactiveHttpInboundEndpoint) handler); + if (mapping != null) { + registerMapping(mapping, handler, HANDLER_METHOD); + } + } + + /** + * Create a {@link RequestMappingInfo} from + * a Spring Integration Reactive HTTP Inbound Endpoint {@link RequestMapping}. + * @see RequestMappingHandlerMapping#getMappingForMethod + */ + private RequestMappingInfo getMappingForEndpoint(ReactiveHttpInboundEndpoint endpoint) { + org.springframework.web.bind.annotation.RequestMapping requestMappingAnnotation = + HttpContextUtils.convertRequestMappingToAnnotation(endpoint.getRequestMapping()); + if (requestMappingAnnotation != null) { + return createRequestMappingInfo(requestMappingAnnotation, getCustomTypeCondition(endpoint.getClass())); + } + else { + return null; + } + } + + @Override + protected CorsConfiguration initCorsConfiguration(Object handler, Method method, RequestMappingInfo mappingInfo) { + CrossOrigin crossOrigin = ((BaseHttpInboundEndpoint) handler).getCrossOrigin(); + if (crossOrigin != null) { + CorsConfiguration config = new CorsConfiguration(); + for (String origin : crossOrigin.getOrigin()) { + config.addAllowedOrigin(origin); + } + for (RequestMethod requestMethod : crossOrigin.getMethod()) { + config.addAllowedMethod(requestMethod.name()); + } + for (String header : crossOrigin.getAllowedHeaders()) { + config.addAllowedHeader(header); + } + for (String header : crossOrigin.getExposedHeaders()) { + config.addExposedHeader(header); + } + if (crossOrigin.getAllowCredentials() != null) { + config.setAllowCredentials(crossOrigin.getAllowCredentials()); + } + if (crossOrigin.getMaxAge() != -1) { + config.setMaxAge(crossOrigin.getMaxAge()); + } + if (CollectionUtils.isEmpty(config.getAllowedMethods())) { + for (RequestMethod allowedMethod : mappingInfo.getMethodsCondition().getMethods()) { + config.addAllowedMethod(allowedMethod.name()); + } + } + if (CollectionUtils.isEmpty(config.getAllowedHeaders())) { + for (NameValueExpression headerExpression : mappingInfo.getHeadersCondition().getExpressions()) { + if (!headerExpression.isNegated()) { + config.addAllowedHeader(headerExpression.getName()); + } + } + } + return config.applyPermitDefaultValues(); + } + return null; + } + + @Override + public void afterPropertiesSet() { + // No-op in favor of onApplicationEvent + } + + /** + * {@link HttpRequestHandlingEndpointSupport}s may depend on auto-created + * {@code requestChannel}s, so MVC Handlers detection should be postponed + * as late as possible. + * @see org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping#afterPropertiesSet() + */ + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (!this.initialized.getAndSet(true)) { + super.afterPropertiesSet(); + } + } + +} diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/RequestMapping.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/RequestMapping.java index d5f25bda3f..3095a3f8b5 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/RequestMapping.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/RequestMapping.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-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. @@ -33,6 +33,8 @@ import org.springframework.web.bind.annotation.RequestMethod; */ public class RequestMapping { + private String name = ""; + private String[] pathPatterns; private HttpMethod[] methods = new HttpMethod[]{HttpMethod.GET, HttpMethod.POST}; @@ -45,6 +47,14 @@ public class RequestMapping { private String[] produces = new String[0]; + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + public void setPathPatterns(String... pathPatterns) { Assert.notEmpty(pathPatterns, "at least one path pattern is required"); this.pathPatterns = pathPatterns; diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java b/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java index 5617180436..00b8863b6b 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/support/HttpContextUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-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. @@ -16,7 +16,13 @@ package org.springframework.integration.http.support; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.core.annotation.AnnotationUtils; import org.springframework.util.ClassUtils; +import org.springframework.util.ObjectUtils; +import org.springframework.web.bind.annotation.RequestMapping; /** * Utility class for accessing HTTP integration components @@ -33,19 +39,36 @@ public final class HttpContextUtils { } /** - * The {@code boolean} flag to indicate if the {@code org.springframework.web.servlet.DispatcherServlet} + * The {@code boolean} flag to indicate if the + * {@code org.springframework.web.servlet.DispatcherServlet} * is present in the CLASSPATH to allow to register the Integration server components, * e.g. {@code IntegrationGraphController}. */ - public static final boolean SERVLET_PRESENT = + public static final boolean WEB_MVC_PRESENT = ClassUtils.isPresent("org.springframework.web.servlet.DispatcherServlet", - HttpContextUtils.class.getClassLoader()); + HttpContextUtils.class.getClassLoader()); /** - * @see org.springframework.integration.http.config.HttpInboundEndpointParser + * The {@code boolean} flag to indicate if the + * {@code org.springframework.web.reactive.result.method.RequestMappingInfo} + * is present in the CLASSPATH to allow to register the Integration server reactive components. + */ + public static final boolean WEB_FLUX_PRESENT = + ClassUtils.isPresent("org.springframework.web.reactive.result.method.RequestMappingInfo", + HttpContextUtils.class.getClassLoader()); + + /** + * The name for the infrastructure + * {@link org.springframework.integration.http.inbound.IntegrationRequestMappingHandlerMapping} bean. */ public static final String HANDLER_MAPPING_BEAN_NAME = "integrationRequestMappingHandlerMapping"; + /** + * The name for the infrastructure + * {@link org.springframework.integration.http.inbound.ReactiveIntegrationRequestMappingHandlerMapping} bean. + */ + public static final String REACTIVE_HANDLER_MAPPING_BEAN_NAME = "reactiveIntegrationRequestMappingHandlerMapping"; + /** * Represents the environment property for the {@code IntegrationGraphController} request mapping path. */ @@ -62,4 +85,30 @@ public final class HttpContextUtils { */ public static final String GRAPH_CONTROLLER_BEAN_NAME = "integrationGraphController"; + /** + * Converts a provided {@link org.springframework.integration.http.inbound.RequestMapping} + * to the Spring Web {@link RequestMapping} annotation. + * @param requestMapping the {@link org.springframework.integration.http.inbound.RequestMapping} to convert. + * @return the {@link RequestMapping} annotation. + * @since 5.0 + */ + public static RequestMapping convertRequestMappingToAnnotation( + org.springframework.integration.http.inbound.RequestMapping requestMapping) { + if (ObjectUtils.isEmpty(requestMapping.getPathPatterns())) { + return null; + } + + Map requestMappingAttributes = new HashMap<>(); + requestMappingAttributes.put("name", requestMapping.getName()); + requestMappingAttributes.put("value", requestMapping.getPathPatterns()); + requestMappingAttributes.put("path", requestMapping.getPathPatterns()); + requestMappingAttributes.put("method", requestMapping.getRequestMethods()); + requestMappingAttributes.put("params", requestMapping.getParams()); + requestMappingAttributes.put("headers", requestMapping.getHeaders()); + requestMappingAttributes.put("consumes", requestMapping.getConsumes()); + requestMappingAttributes.put("produces", requestMapping.getProduces()); + + return AnnotationUtils.synthesizeAnnotation(requestMappingAttributes, RequestMapping.class, null); + } + } 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 c43b371f3f..23c646c65a 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java @@ -16,6 +16,9 @@ package org.springframework.integration.http.dsl; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.httpBasic; import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; @@ -27,11 +30,13 @@ import java.util.List; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.ResolvableType; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; @@ -44,7 +49,9 @@ import org.springframework.integration.http.outbound.HttpRequestExecutingMessage import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler; import org.springframework.integration.security.channel.ChannelSecurityInterceptor; import org.springframework.integration.security.channel.SecuredChannel; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; import org.springframework.security.access.AccessDecisionManager; import org.springframework.security.access.vote.AffirmativeBased; import org.springframework.security.access.vote.RoleVoter; @@ -57,16 +64,19 @@ import org.springframework.test.context.junit4.SpringRunner; import org.springframework.test.context.web.WebAppConfiguration; import org.springframework.test.web.client.MockMvcClientHttpRequestFactory; import org.springframework.test.web.reactive.server.HttpHandlerConnector; +import org.springframework.test.web.reactive.server.WebTestClient; import org.springframework.test.web.servlet.MockMvc; import org.springframework.test.web.servlet.setup.MockMvcBuilders; import org.springframework.util.MultiValueMap; import org.springframework.web.client.RestTemplate; import org.springframework.web.context.WebApplicationContext; +import org.springframework.web.reactive.config.EnableWebFlux; import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.servlet.config.annotation.EnableWebMvc; import org.springframework.web.util.UriComponentsBuilder; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * @author Artem Bilan @@ -90,12 +100,18 @@ public class HttpDslTests { private MockMvc mockMvc; + private WebTestClient webTestClient; + @Before public void setup() { this.mockMvc = MockMvcBuilders.webAppContextSetup(this.wac) .apply(springSecurity()) .build(); + + this.webTestClient = + WebTestClient.bindToApplicationContext(this.wac) + .build(); } @@ -140,9 +156,45 @@ public class HttpDslTests { .string("FOO")); } + @Autowired + private PollableChannel storeChannel; + + + @Test + @SuppressWarnings("unchecked") + public void testHttpReactivePost() { + this.webTestClient.post().uri("/reactivePost") + .body(Flux.just("foo", "bar", "baz"), String.class) + .exchange() + .expectStatus().isAccepted(); + + Message store = this.storeChannel.receive(10_000); + assertNotNull(store); + assertThat(store.getPayload(), instanceOf(Flux.class)); + + StepVerifier + .create((Publisher) store.getPayload()) + .expectNext("foo", "bar", "baz") + .verifyComplete(); + + } + + @Test + public void testSse() { + Flux responseBody = + this.webTestClient.get().uri("/sse") + .exchange() + .returnResult(String.class) + .getResponseBody(); + + StepVerifier + .create(responseBody) + .expectNext("foo", "bar", "baz") + .verifyComplete(); + } @Configuration - @EnableWebMvc + @EnableWebFlux @EnableWebSecurity @EnableIntegration public static class ContextConfiguration extends WebSecurityConfigurerAdapter { @@ -211,6 +263,26 @@ public class HttpDslTests { .get(); } + @Bean + public IntegrationFlow httpReactiveInboundChannelAdapterFlow() { + return IntegrationFlows + .from(Http.inboundReactiveChannelAdapter("/reactivePost") + .requestMapping(m -> m.methods(HttpMethod.POST)) + .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class)) + .statusCodeFunction(m -> HttpStatus.ACCEPTED)) + .channel(c -> c.queue("storeChannel")) + .get(); + } + + @Bean + public IntegrationFlow sseFlow() { + return IntegrationFlows + .from(Http.inboundReactiveGateway("/sse") + .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) + .handle((p, h) -> Flux.just("foo", "bar", "baz")) + .get(); + } + @Bean public AccessDecisionManager accessDecisionManager() { return new AffirmativeBased(Collections.singletonList(new RoleVoter())); diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGatewayTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGatewayTests.java index b6a9bee734..72f97259c9 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGatewayTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGatewayTests.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. @@ -64,6 +64,7 @@ import org.springframework.util.SerializationUtils; * @author Gunnar Hillert * @author Artem Bilan * @author Biju Kunjummen + * * @since 2.0 */ public class HttpRequestHandlingMessagingGatewayTests extends AbstractHttpInboundTests { @@ -96,7 +97,7 @@ public class HttpRequestHandlingMessagingGatewayTests extends AbstractHttpInboun QueueChannel requestChannel = new QueueChannel(); HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(false); gateway.setBeanFactory(mock(BeanFactory.class)); - gateway.setRequestPayloadType(String.class); + gateway.setRequestPayloadTypeClass(String.class); gateway.setRequestChannel(requestChannel); gateway.afterPropertiesSet(); gateway.start(); @@ -134,7 +135,7 @@ public class HttpRequestHandlingMessagingGatewayTests extends AbstractHttpInboun HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(true); gateway.setStatusCodeExpression(new LiteralExpression("foo")); gateway.setBeanFactory(mock(BeanFactory.class)); - gateway.setRequestPayloadType(String.class); + gateway.setRequestPayloadTypeClass(String.class); gateway.setRequestChannel(requestChannel); gateway.afterPropertiesSet(); gateway.start(); @@ -162,7 +163,7 @@ public class HttpRequestHandlingMessagingGatewayTests extends AbstractHttpInboun }); HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(true); gateway.setBeanFactory(mock(BeanFactory.class)); - gateway.setRequestPayloadType(String.class); + gateway.setRequestPayloadTypeClass(String.class); gateway.setRequestChannel(requestChannel); gateway.afterPropertiesSet(); gateway.start(); @@ -238,7 +239,7 @@ public class HttpRequestHandlingMessagingGatewayTests extends AbstractHttpInboun QueueChannel channel = new QueueChannel(); HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(false); gateway.setBeanFactory(mock(BeanFactory.class)); - gateway.setRequestPayloadType(TestBean.class); + gateway.setRequestPayloadTypeClass(TestBean.class); gateway.setRequestChannel(channel); List> converters = new ArrayList>(); diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/MultipartAsRawByteArrayTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/MultipartAsRawByteArrayTests.java index 1071d6dd7e..9a6e9a86b8 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/MultipartAsRawByteArrayTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/MultipartAsRawByteArrayTests.java @@ -47,6 +47,7 @@ import org.springframework.web.context.request.RequestContextHolder; /** * @author Gary Russell * @author Artem Bilan + * * @since 4.2 */ public class MultipartAsRawByteArrayTests { @@ -61,7 +62,7 @@ public class MultipartAsRawByteArrayTests { QueueChannel requestChannel = new QueueChannel(); gw.setRequestChannel(requestChannel); gw.setBeanFactory(mock(BeanFactory.class)); - gw.setRequestPayloadType(byte[].class); + gw.setRequestPayloadTypeClass(byte[].class); gw.afterPropertiesSet(); gw.start(); diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java new file mode 100644 index 0000000000..2c631fb796 --- /dev/null +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java @@ -0,0 +1,170 @@ +/* + * 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.inbound; + +import java.util.Objects; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.web.reactive.server.WebTestClient; +import org.springframework.web.reactive.config.EnableWebFlux; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import reactor.core.publisher.Flux; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class ReactiveHttpInboundEndpointTests { + + @Autowired + private WebTestClient webTestClient; + + @Autowired + private ReactiveHttpInboundEndpoint simpleInboundEndpoint; + + @Test + public void testSimpleGet() { + this.webTestClient.get().uri("/test") + .exchange() + .expectStatus().isOk() + .expectBody(String.class).isEqualTo("It works!"); + + this.simpleInboundEndpoint.stop(); + + this.webTestClient.get().uri("/test") + .exchange() + .expectStatus().isEqualTo(HttpStatus.SERVICE_UNAVAILABLE) + .expectBody(String.class).isEqualTo("Endpoint is stopped"); + } + + @Test + public void testJsonResult() { + this.webTestClient.get().uri("/persons") + .accept(MediaType.APPLICATION_JSON_UTF8) + .exchange() + .expectStatus().isOk() + .expectBody() + .jsonPath("$[0].name").isEqualTo("Jane") + .jsonPath("$[1].name").isEqualTo("Jason") + .jsonPath("$[2].name").isEqualTo("John"); + } + + @Configuration + @EnableWebFlux + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public WebTestClient webTestClient(ApplicationContext applicationContext) { + return WebTestClient.bindToApplicationContext(applicationContext).build(); + } + + @Bean + public ReactiveHttpInboundEndpoint simpleInboundEndpoint() { + ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/test"); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannelName("serviceChannel"); + return endpoint; + } + + @ServiceActivator(inputChannel = "serviceChannel") + String service() { + return "It works!"; + } + + @Bean + public ReactiveHttpInboundEndpoint jsonInboundEndpoint() { + ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/persons"); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannel(fluxResultChannel()); + return endpoint; + } + + @Bean + public MessageChannel fluxResultChannel() { + return new FluxMessageChannel(); + } + + @ServiceActivator(inputChannel = "fluxResultChannel") + Flux getPersons() { + return Flux.just(new Person("Jane"), new Person("Jason"), new Person("John")); + } + + } + + + static class Person { + + private final String name; + + @JsonCreator + Person(@JsonProperty("name") String name) { + this.name = name; + } + + public String getName() { + return this.name; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Person person = (Person) o; + return Objects.equals(this.name, person.name); + } + + @Override + public int hashCode() { + return getName().hashCode(); + } + + @Override + public String toString() { + return "Person[name='" + this.name + "']"; + } + + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java index b597b23f0c..d9f833f84b 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java @@ -45,6 +45,7 @@ import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; /** * @author Artem Bilan * @author Gary Russell + * * @since 4.3.1 */ public class JdbcLockRegistryLeaderInitiatorTests { @@ -155,15 +156,16 @@ public class JdbcLockRegistryLeaderInitiatorTests { assertThat(initiator1.getContext().isLeader(), is(true)); assertThat(initiator2.getContext().isLeader(), is(false)); + // Stop second initiator, so the first one will be leader even after yield initiator2.stop(); - CountDownLatch revoked11 = new CountDownLatch(1); - initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11)); + CountDownLatch granted11 = new CountDownLatch(1); + initiator1.setLeaderEventPublisher(new CountingPublisher(granted11)); initiator1.getContext().yield(); - assertThat(revoked11.await(20, TimeUnit.SECONDS), is(true)); - assertThat(initiator1.getContext().isLeader(), is(false)); + assertThat(granted11.await(20, TimeUnit.SECONDS), is(true)); + assertThat(initiator1.getContext().isLeader(), is(true)); initiator1.stop(); } diff --git a/src/reference/asciidoc/http.adoc b/src/reference/asciidoc/http.adoc index ed17e08fc5..a04caa5a3f 100644 --- a/src/reference/asciidoc/http.adoc +++ b/src/reference/asciidoc/http.adoc @@ -122,6 +122,59 @@ This also shows how to customize the HTTP methods accepted by the gateway, which The reply message will be available in the Model map. The key that is used for that map entry by default is 'reply', but this can be overridden by setting the 'replyKey' property on the endpoint's configuration. +=== WebFlux Server Side support + +Starting with _version 5.0_, the `ReactiveHttpInboundEndpoint`, http://docs.spring.io/spring/docs/5.0.0.RC3/spring-framework-reference/web.html#web-reactive[WebFlux] `WebHandler`, implementation is provided. +This component is similar to the MVC-based `HttpRequestHandlingEndpointSupport` with which it shares some common options via the newly extracted `BaseHttpInboundEndpoint`. +Instead of MVC, it is used in the Spring WebFlux Reactive environment. +A simple sample for explanation: + +[source,java] +---- +@Configuration +@EnableWebFlux +@EnableIntegration +public class ReactiveHttpConfiguration { + + @Bean + public ReactiveHttpInboundEndpoint simpleInboundEndpoint() { + ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/test"); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannelName("serviceChannel"); + return endpoint; + } + + @ServiceActivator(inputChannel = "serviceChannel") + String service() { + return "It works!"; + } + +} +---- + +As can be seen, the configuration is similar to the `HttpRequestHandlingEndpointSupport` mentioned above, except that we use `@EnableWebFlux` to add the WebFlux infrastructure to our integration application. +Also, the `ReactiveHttpInboundEndpoint` performs `sendAndReceive` operation to the downstream flow using back-pressure, on demand based capabilities, provided by the reactive HTTP server implementation. + +NOTE: The reply part is non-blocking as well and based on the internal `FutureReplyChannel` which is flat-mapped to a reply `Mono` for on demand resolution. + +The `ReactiveHttpInboundEndpoint` can be configured with a custom `ServerCodecConfigurer`, `RequestedContentTypeResolver` and even a `ReactiveAdapterRegistry`. +The latter provides a mechanism where we can return a reply as any reactive type - Reactor `Flux`, RxJava `Observable`, `Flowable` etc. +This way, we can simply implement https://en.wikipedia.org/wiki/Server-sent_events[Server Sent Events] scenarios with Spring Integration components: + +[source,java] +---- +@Bean +public IntegrationFlow sseFlow() { + return IntegrationFlows + .from(Http.inboundReactiveGateway("/sse") + .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) + .handle((p, h) -> Flux.just("foo", "bar", "baz")) + .get(); +} +---- + [[http-outbound]] === Http Outbound Components ==== HttpRequestExecutingMessageHandler @@ -193,8 +246,8 @@ Of course, this can be an abstract class, or even an interface (such as `java.io ==== ReactiveHttpRequestExecutingMessageHandler -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: +The `ReactiveHttpRequestExecutingMessageHandler` (starting with _version 5.0_) implementation is very similar to `HttpRequestExecutingMessageHandler`, using a `WebClient` from the Spring Framework WebFlux module. +To configure it, define a bean like this: [source,xml] ---- @@ -686,6 +739,26 @@ public RequestMapping mapping() { requestMapping.setMethods(HttpMethod.POST); return requestMapping; } + +@Bean +public ReactiveHttpInboundEndpoint jsonInboundEndpoint() { + ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/persons"); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannel(fluxResultChannel()); + return endpoint; +} + +@Bean +public MessageChannel fluxResultChannel() { + return new FluxMessageChannel(); +} + +@ServiceActivator(inputChannel = "fluxResultChannel") +Flux getPersons() { + return Flux.just(new Person("Jane"), new Person("Jason"), new Person("John")); +} ---- .Inbound Gateway Using the Java DSL @@ -699,6 +772,17 @@ public IntegrationFlow inbound() { .channel("httpRequest") .get(); } + +@Bean +public IntegrationFlow httpReactiveInboundChannelAdapterFlow() { + return IntegrationFlows + .from(Http.inboundReactiveChannelAdapter("/reactivePost") + .requestMapping(m -> m.methods(HttpMethod.POST)) + .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class)) + .statusCodeFunction(m -> HttpStatus.ACCEPTED)) + .channel(c -> c.queue("storeChannel")) + .get(); +} ---- .Outbound Gateway Using Java Configuration @@ -713,6 +797,16 @@ public HttpRequestExecutingMessageHandler outbound() { handler.setExpectedResponseType(String.class); return handler; } + +@ServiceActivator(inputChannel = "reactiveHttpOutRequest") +@Bean +public ReactiveHttpRequestExecutingMessageHandler reactiveOutbound(WebClient client) { + ReactiveHttpRequestExecutingMessageHandler handler = + new ReactiveHttpRequestExecutingMessageHandler("http://localhost:8080/foo", client); + handler.setHttpMethod(HttpMethod.POST); + handler.setExpectedResponseType(String.class); + return handler; +} ---- .Outbound Gateway Using the Java DSL @@ -726,6 +820,18 @@ public IntegrationFlow outbound() { .expectedResponseType(String.class)) .get(); } + +@Bean +public IntegrationFlow outboundReactive() { + return f -> f + .handle(Http.>outboundReactiveGateway(m -> + UriComponentsBuilder.fromUriString("http://localhost:8080/foo") + .queryParams(m.getPayload()) + .build() + .toUri()) + .httpMethod(HttpMethod.GET) + .expectedResponseType(String.class)); +} ---- [[http-timeout]] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 50a4bb878a..731a441a33 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -29,11 +29,11 @@ The new `MongoDbOutboundGateway` allows you to make queries to the database on d See <> for more information. -==== HTTP Reactive Outbound Gateway and Channel Adapter +==== HTTP Reactive Inbound and Outbound Gateways and Channel Adapters -The new `ReactiveHttpRequestExecutingMessageHandler` adds support for WebFlux `WebClient` for outbound channel adapter and gateway. +The new `ReactiveHttpInboundEndpoint` and `ReactiveHttpRequestExecutingMessageHandler` add support for Spring WebFlux Framework gateways and channel adapters. -See <> for more information. +See <> for more information. ==== Content Type Conversion