Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

WebFlux improvements

Browse Source 
* add `BodyExtractor` support for Outbound part
* add `ClientHttpResponseBodyExtractor` as identity function
* add XML configuration for the Inbound part
* document `BodyExtractor` and Inbound XML support

Rename `replyToFlux` property to the `replyPayloadToFlux`

Doc Polishing
This commit is contained in:
Artem Bilan
2017-12-20 11:11:09 -05:00
committed by Gary Russell
parent 1e1346dc94
commit b5ec98f025
16 changed files with 960 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxInboundEndpointParser.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.webflux.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.http.config.HttpInboundEndpointParser;
import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
/**
* @author Artem Bilan
*
* @since 5.0.1
*/
public class WebFluxInboundEndpointParser extends HttpInboundEndpointParser {
public WebFluxInboundEndpointParser(boolean expectReply) {
super(expectReply);
}
@Override
protected Class<?> getBeanClass(Element element) {
return WebFluxInboundEndpoint.class;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
super.doParse(element, parserContext, builder);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "codec-configurer");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "requested-content-type-resolver");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reactive-adapter-registry");
}
}

4
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxNamespaceHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -28,6 +28,8 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
public class WebFluxNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new WebFluxInboundEndpointParser(false));
registerBeanDefinitionParser("inbound-gateway", new WebFluxInboundEndpointParser(true));
registerBeanDefinitionParser("outbound-channel-adapter", new WebFluxOutboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-gateway", new WebFluxOutboundGatewayParser());
}

5
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -47,7 +47,8 @@ public class WebFluxOutboundGatewayParser extends HttpOutboundGatewayParser {
.addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef));
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-to-flux");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-payload-to-flux");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "body-extractor");
return builder;
}

31
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxMessageHandlerSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -20,11 +20,16 @@ import java.net.URI;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.http.dsl.BaseHttpMessageHandlerSpec;
import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler;
import org.springframework.web.reactive.function.BodyExtractor;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* The {@link BaseHttpMessageHandlerSpec} implementation for the {@link WebFluxRequestExecutingMessageHandler}.
*
@@ -53,6 +58,30 @@ public class WebFluxMessageHandlerSpec
this.webClient = webClient;
}
/**
* The boolean flag to identify if the reply payload should be as a {@link Flux} from the response body
* or as resolved value from the {@link Mono} of the response body.
* Defaults to {@code false} - simple value is pushed downstream.
* Makes sense when {@code expectedResponseType} is configured.
* @param replyPayloadToFlux represent reply payload as a {@link Flux} or as a value from the {@link Mono}.
* @since 5.0.1
* @see WebFluxRequestExecutingMessageHandler#setReplyPayloadToFlux(boolean)
*/
public void replyPayloadToFlux(boolean replyPayloadToFlux) {
this.target.setReplyPayloadToFlux(replyPayloadToFlux);
}
/**
* Specify a {@link BodyExtractor} as an alternative to the {@code expectedResponseType}
* to allow to get low-level access to the received {@link ClientHttpResponse}.
* @param bodyExtractor the {@link BodyExtractor} to use.
* @since 5.0.1
* @see WebFluxRequestExecutingMessageHandler#setBodyExtractor(BodyExtractor)
*/
public void bodyExtractor(BodyExtractor<?, ClientHttpResponse> bodyExtractor) {
this.target.setBodyExtractor(bodyExtractor);
}
@Override
protected boolean isClientSet() {
return this.webClient != null;

4
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -126,7 +126,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
@Override
public String getComponentType() {
return super.getComponentType().replaceFirst("(http:)", "$1webflux-");
return super.getComponentType().replaceFirst("http", "webflux");
}
@Override

36
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ReactiveHttpInputMessage;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.http.outbound.AbstractHttpRequestExecutingMessageHandler;
import org.springframework.messaging.Message;
@@ -62,7 +63,9 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
private final WebClient webClient;
private boolean replyToFlux;
private boolean replyPayloadToFlux;
private BodyExtractor<?, ClientHttpResponse> bodyExtractor;
/**
* Create a handler that will send requests to the provided URI.
@@ -120,13 +123,25 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
* or as resolved value from the {@link Mono} of the response body.
* Defaults to {@code false} - simple value is pushed downstream.
* Makes sense when {@code expectedResponseType} is configured.
* @param replyToFlux represent reply payload as a {@link Flux} or as a value from the {@link Mono}.
* @param replyPayloadToFlux represent reply payload as a {@link Flux} or as a value from the {@link Mono}.
* @since 5.0.1
* @see #setExpectedResponseType(Class)
* @see #setExpectedResponseTypeExpression(Expression)
*/
public void setReplyToFlux(boolean replyToFlux) {
this.replyToFlux = replyToFlux;
public void setReplyPayloadToFlux(boolean replyPayloadToFlux) {
this.replyPayloadToFlux = replyPayloadToFlux;
}
/**
* Specify a {@link BodyExtractor} as an alternative to the {@code expectedResponseType}
* to allow to get low-level access to the received {@link ClientHttpResponse}.
* @param bodyExtractor the {@link BodyExtractor} to use.
* @since 5.0.1
* @see #setExpectedResponseType(Class)
* @see #setExpectedResponseTypeExpression(Expression)
*/
public void setBodyExtractor(BodyExtractor<?, ClientHttpResponse> bodyExtractor) {
this.bodyExtractor = bodyExtractor;
}
@Override
@@ -191,7 +206,7 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
Mono<?> bodyMono;
if (expectedResponseType != null) {
if (this.replyToFlux) {
if (this.replyPayloadToFlux) {
BodyExtractor<? extends Flux<?>, ReactiveHttpInputMessage> extractor;
if (expectedResponseType instanceof ParameterizedTypeReference<?>) {
extractor = BodyExtractors.toFlux(
@@ -215,6 +230,15 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
bodyMono = response.body(extractor);
}
}
else if (this.bodyExtractor != null) {
Object body = response.body(this.bodyExtractor);
if (body instanceof Mono) {
bodyMono = (Mono<?>) body;
}
else {
bodyMono = Mono.just(body);
}
}
else {
bodyMono = Mono.empty();
}

37
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/ClientHttpResponseBodyExtractor.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.webflux.support;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.web.reactive.function.BodyExtractor;
/**
* The {@link BodyExtractor} identity function implementation
* which just returns the provided {@link ClientHttpResponse}.
*
* @author Artem Bilan
*
* @since 5.0.1
*/
public class ClientHttpResponseBodyExtractor implements BodyExtractor<ClientHttpResponse, ClientHttpResponse> {
@Override
public ClientHttpResponse extract(ClientHttpResponse response, Context context) {
return response;
}
}

325
spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.0.xsd
View File

@@ -20,6 +20,315 @@
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures a Message Producing Endpoint for the
'org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint' that
receives HTTP requests reactive manner and doesn't produce responses.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element name="request-mapping" type="int-http:requestMappingType" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
Defines configuration for org.springframework.integration.http.inbound.RequestMapping
as RESTFul attributes for Spring Integration HTTP Inbound Endpoints.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="cross-origin" type="int-http:crossOriginType" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
Marks this endpoint as permitting cross origin requests (CORS).
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="header" type="int-http:headerType" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation>
Specifies a Message header as a result of expression evaluation
against ServletRequest and URI Variables.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Maximum amount of time in milliseconds to wait when sending
a message to the channel if such channel may block.
For example, a Queue Channel can block until space
is available if its maximum capacity has been reached.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="status-code-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A SpEL expression that resolves to an 'HttpStatus' code when rendering a response.
The expression must return an object which can be converted to a
'org.springframework.http.HttpStatus' enum value.
The 'evaluationContext' has a 'BeanResolver' but no variables, so the usage of this attribute
is somewhat limited.
An example might be to resolve, at runtime, some scoped Bean that returns an
'HttpStatus' value, or use a literal expression e.g. "201".
By default 'status-code-expression' is null, meaning that the default '200 OK' response status
will be returned.
The 'http:inbound-gateway' resolves the 'status code' from the 'http_statusCode' header of the reply
Message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="inboundCommonAttributes" />
</xsd:complexType>
</xsd:element>
<xsd:element name="inbound-gateway">
<xsd:annotation>
<xsd:documentation>
Configures a Messaging Gateway Endpoint for the
'org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint' that
receives HTTP requests and produces responses reactive manner.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="int-http:gatewayType">
<xsd:sequence>
<xsd:element name="request-mapping" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
Defines configuration for org.springframework.integration.http.inbound.RequestMapping
as RESTFul attributes for Spring Integration HTTP Inbound Endpoints.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="int-http:requestMappingType">
<xsd:attribute name="produces" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The producible media types of the mapped request, narrowing the primary mapping.
The format is a sequence of media types ("text/plain", "application/*),
with a request only mapped if the Accept matches one of these media types.
Expressions can be negated by using the "!" operator, as in "!text/plain", which matches
all requests with a Accept other than "text/plain".
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="cross-origin" type="int-http:crossOriginType" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
Marks this endpoint as permitting cross origin requests (CORS).
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="header" type="int-http:headerType" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation>
Specifies a Message header as a result of expression evaluation
against ServletRequest and URI Variables.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="request-channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The receiving Message Channel of this endpoint.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="inboundCommonAttributes" />
<xsd:attribute name="extract-reply-payload" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-response-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the HttpHeaders of the HTTP response.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
The String "HTTP_RESPONSE_HEADERS" will match against any of the standard HTTP Response headers.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Used to set the sendTimeout on the underlying MessagingTemplate instance
(org.springframework.integration.core.MessagingTemplate) for sending messages
to the request channel. If not specified this property will default to "1000"
(1 second).
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Used to set the receiveTimeout on the underlying MessagingTemplate instance
(org.springframework.integration.core.MessagingTemplate) for receiving messages
from the reply channel. If not specified this property will default to "1000"
(1 second).
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout-status-code-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A SpEL expression that resolves to an 'HttpStatus' code when rendering a response after
a 'reply-timeout'.
The expression must return an object which can be converted to a
'org.springframework.http.HttpStatus' enum value.
The 'evaluationContext' has a 'BeanResolver' but no variables, so the usage of this attribute
is somewhat limited.
An example might be to resolve, at runtime, some scoped Bean that returns an
'HttpStatus' value, or use a literal expression e.g. "504".
By default 'status-code-expression' is null, meaning that the default
'500 Internal Server Error' response status will be returned after a timeout.
When a timeout is not encountered,
the 'http:inbound-gateway' resolves the 'status code' from the 'http_statusCode' header of the reply
Message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:attributeGroup name="inboundCommonAttributes">
<xsd:attribute name="path" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Comma-separated URI paths (e.g., /orderId/{order}).
Ant-style path patterns are also supported (e.g. /myPath/*.do).
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="supported-methods">
<xsd:annotation>
<xsd:documentation>
Comma-separated HTTP Method names. Determines which types of Request are
allowed with this Endpoint.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="int-http:httpMethodEnumeration xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="request-payload-type" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Target type for payload that is the conversion result of the request.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="payload-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Allows you to specify SpEL expression to construct a Message payload
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Specifies a reference to org.springframework.integration.mapping.HeaderMapper
implementation bean. Only one of 'header-mapper' or 'mapped-request-headers' attributes
can be provided.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of HttpHeaders to be mapped from the HTTP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
The String "HTTP_REQUEST_HEADERS" will match against any of the standard HTTP Request headers.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The MessagingGateway's 'error-channel' where to send an ErrorMessage in case
of Exception is caused from original message flow.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="codec-configurer" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.http.codec.ServerCodecConfigurer" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A 'ServerCodecConfigurer' for the request readers and response writers.
By default the 'ServerCodecConfigurer#create()' factory is used.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="requested-content-type-resolver" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.web.reactive.accept.RequestedContentTypeResolver" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A 'RequestedContentTypeResolver' strategy to resolve the requested media types
for a 'ServerWebExchange'.
A 'HeaderContentTypeResolver' is used by default.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reactive-adapter-registry" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.ReactiveAdapterRegistry" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A 'ReactiveAdapterRegistry' registry of adapters to adapt a
Reactive Streams 'Publisher' to/from.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:attributeGroup>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
@@ -177,7 +486,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-to-flux" default="false">
<xsd:attribute name="reply-payload-to-flux" default="false">
<xsd:annotation>
<xsd:documentation>
When set to "true", the response is converted to the Flux which is sent as a reply
@@ -190,6 +499,20 @@
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="body-extractor" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.web.reactive.function.BodyExtractor"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A reference to an org.springframework.web.reactive.function.BodyExtractor bean
which is used to to extract the content from the response.
Must be used instead of 'expected-response-type(-expression)'
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

48
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests-context.xml Normal file
View File

@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans
xmlns="http://www.springframework.org/schema/integration/webflux"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration/webflux http://www.springframework.org/schema/integration/webflux/spring-integration-webflux.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<si:channel id="requests"/>
<inbound-channel-adapter id="reactiveMinimalConfig" channel="requests"/>
<inbound-channel-adapter id="reactiveFullConfig" channel="requests"
path="test1"
auto-startup="false"
phase="101"
request-payload-type="byte[]"
error-channel="errorChannel"
payload-expression="payload"
supported-methods="PUT"
status-code-expression="'202'"
header-mapper="headerMapper"
codec-configurer="codecConfigurer"
reactive-adapter-registry="reactiveAdapterRegistry"
requested-content-type-resolver="requestedContentTypeResolver">
<request-mapping headers="foo"/>
<cross-origin origin="foo"
method="PUT"/>
<header name="foo" expression="'foo'"/>
</inbound-channel-adapter>
<beans:bean id="headerMapper" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.integration.mapping.HeaderMapper"/>
</beans:bean>
<beans:bean id="codecConfigurer" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.http.codec.ServerCodecConfigurer"/>
</beans:bean>
<beans:bean id="requestedContentTypeResolver" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.web.reactive.accept.RequestedContentTypeResolver"/>
</beans:bean>
<beans:bean id="reactiveAdapterRegistry" class="org.springframework.core.ReactiveAdapterRegistry"/>
</beans:beans>

134
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java Normal file
View File

@@ -0,0 +1,134 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.webflux.config;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.ResolvableType;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.integration.http.inbound.CrossOrigin;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
/**
* @author Artem Bilan
*
* @since 5.0.1
*/
@RunWith(SpringRunner.class)
@DirtiesContext
public class WebFluxInboundChannelAdapterParserTests {
@Autowired
@Qualifier("reactiveMinimalConfig")
private WebFluxInboundEndpoint reactiveMinimalConfig;
@Autowired
@Qualifier("reactiveFullConfig")
private WebFluxInboundEndpoint reactiveFullConfig;
@Autowired
private MessageChannel requests;
@Autowired
private HeaderMapper<?> headerMapper;
@Autowired
private ServerCodecConfigurer serverCodecConfigurer;
@Autowired
private RequestedContentTypeResolver requestedContentTypeResolver;
@Autowired
private ReactiveAdapterRegistry reactiveAdapterRegistry;
@Test
public void reactiveMinimalConfig() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveMinimalConfig);
assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel"));
assertTrue((boolean) endpointAccessor.getPropertyValue("autoStartup"));
assertFalse((boolean) endpointAccessor.getPropertyValue("expectReply"));
assertNull(endpointAccessor.getPropertyValue("statusCodeExpression"));
assertNull(endpointAccessor.getPropertyValue("payloadExpression"));
assertNull(endpointAccessor.getPropertyValue("headerExpressions"));
assertNull(endpointAccessor.getPropertyValue("crossOrigin"));
assertNull(endpointAccessor.getPropertyValue("requestPayloadType"));
assertNotSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper"));
assertNotSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer"));
assertNotSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver"));
assertNotSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry"));
}
@Test
@SuppressWarnings("unchecked")
public void reactiveFullConfig() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfig);
assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel"));
assertNotNull(endpointAccessor.getPropertyValue("errorChannel"));
assertFalse((boolean) endpointAccessor.getPropertyValue("autoStartup"));
assertEquals(101, endpointAccessor.getPropertyValue("phase"));
assertFalse((boolean) endpointAccessor.getPropertyValue("expectReply"));
assertEquals("'202'",
((SpelExpression) endpointAccessor.getPropertyValue("statusCodeExpression")).getExpressionString());
assertEquals("payload",
((SpelExpression) endpointAccessor.getPropertyValue("payloadExpression")).getExpressionString());
Map<String, Expression> headerExpressions =
(Map<String, Expression>) endpointAccessor.getPropertyValue("headerExpressions");
assertTrue(headerExpressions.containsKey("foo"));
assertEquals("foo", headerExpressions.get("foo").getValue());
CrossOrigin crossOrigin = (CrossOrigin) endpointAccessor.getPropertyValue("crossOrigin");
assertNotNull(crossOrigin);
assertArrayEquals(new String[] { "foo" }, crossOrigin.getOrigin());
assertEquals(ResolvableType.forClass(byte[].class), endpointAccessor.getPropertyValue("requestPayloadType"));
assertSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper"));
assertSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer"));
assertSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver"));
assertSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry"));
}
}

48
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests-context.xml Normal file
View File

@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans
xmlns="http://www.springframework.org/schema/integration/webflux"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration/webflux http://www.springframework.org/schema/integration/webflux/spring-integration-webflux.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<si:channel id="requests"/>
<inbound-gateway id="reactiveMinimalConfig" request-channel="requests"/>
<inbound-gateway id="reactiveFullConfig" request-channel="requests"
path="test1"
auto-startup="false"
phase="101"
request-payload-type="byte[]"
error-channel="errorChannel"
payload-expression="payload"
supported-methods="PUT"
reply-timeout-status-code-expression="'504'"
header-mapper="headerMapper"
codec-configurer="codecConfigurer"
reactive-adapter-registry="reactiveAdapterRegistry"
requested-content-type-resolver="requestedContentTypeResolver">
<request-mapping headers="foo"/>
<cross-origin origin="foo"
method="PUT"/>
<header name="foo" expression="'foo'"/>
</inbound-gateway>
<beans:bean id="headerMapper" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.integration.mapping.HeaderMapper"/>
</beans:bean>
<beans:bean id="codecConfigurer" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.http.codec.ServerCodecConfigurer"/>
</beans:bean>
<beans:bean id="requestedContentTypeResolver" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.web.reactive.accept.RequestedContentTypeResolver"/>
</beans:bean>
<beans:bean id="reactiveAdapterRegistry" class="org.springframework.core.ReactiveAdapterRegistry"/>
</beans:beans>

134
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java Normal file
View File

@@ -0,0 +1,134 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.webflux.config;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.ResolvableType;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.integration.http.inbound.CrossOrigin;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
/**
* @author Artem Bilan
*
* @since 5.0.1
*/
@RunWith(SpringRunner.class)
@DirtiesContext
public class WebFluxInboundGatewayParserTests {
@Autowired
@Qualifier("reactiveMinimalConfig")
private WebFluxInboundEndpoint reactiveMinimalConfig;
@Autowired
@Qualifier("reactiveFullConfig")
private WebFluxInboundEndpoint reactiveFullConfig;
@Autowired
private MessageChannel requests;
@Autowired
private HeaderMapper<?> headerMapper;
@Autowired
private ServerCodecConfigurer serverCodecConfigurer;
@Autowired
private RequestedContentTypeResolver requestedContentTypeResolver;
@Autowired
private ReactiveAdapterRegistry reactiveAdapterRegistry;
@Test
public void reactiveMinimalConfig() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveMinimalConfig);
assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel"));
assertTrue((boolean) endpointAccessor.getPropertyValue("autoStartup"));
assertTrue((boolean) endpointAccessor.getPropertyValue("expectReply"));
assertNull(endpointAccessor.getPropertyValue("statusCodeExpression"));
assertNull(endpointAccessor.getPropertyValue("payloadExpression"));
assertNull(endpointAccessor.getPropertyValue("headerExpressions"));
assertNull(endpointAccessor.getPropertyValue("crossOrigin"));
assertNull(endpointAccessor.getPropertyValue("requestPayloadType"));
assertNotSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper"));
assertNotSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer"));
assertNotSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver"));
assertNotSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry"));
}
@Test
@SuppressWarnings("unchecked")
public void reactiveFullConfig() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfig);
assertSame(this.requests, endpointAccessor.getPropertyValue("requestChannel"));
assertNotNull(endpointAccessor.getPropertyValue("errorChannel"));
assertFalse((boolean) endpointAccessor.getPropertyValue("autoStartup"));
assertEquals(101, endpointAccessor.getPropertyValue("phase"));
assertTrue((boolean) endpointAccessor.getPropertyValue("expectReply"));
assertEquals("'504'",
((SpelExpression) endpointAccessor.getPropertyValue("statusCodeExpression")).getExpressionString());
assertEquals("payload",
((SpelExpression) endpointAccessor.getPropertyValue("payloadExpression")).getExpressionString());
Map<String, Expression> headerExpressions =
(Map<String, Expression>) endpointAccessor.getPropertyValue("headerExpressions");
assertTrue(headerExpressions.containsKey("foo"));
assertEquals("foo", headerExpressions.get("foo").getValue());
CrossOrigin crossOrigin = (CrossOrigin) endpointAccessor.getPropertyValue("crossOrigin");
assertNotNull(crossOrigin);
assertArrayEquals(new String[] { "foo" }, crossOrigin.getOrigin());
assertEquals(ResolvableType.forClass(byte[].class), endpointAccessor.getPropertyValue("requestPayloadType"));
assertSame(this.headerMapper, endpointAccessor.getPropertyValue("headerMapper"));
assertSame(this.serverCodecConfigurer, endpointAccessor.getPropertyValue("codecConfigurer"));
assertSame(this.requestedContentTypeResolver, endpointAccessor.getPropertyValue("requestedContentTypeResolver"));
assertSame(this.reactiveAdapterRegistry, endpointAccessor.getPropertyValue("adapterRegistry"));
}
}

6
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml
View File

@@ -36,8 +36,12 @@
order="77"
auto-startup="false"
transfer-cookies="true"
reply-to-flux="true">
reply-payload-to-flux="true"
body-extractor="bodyExtractor">
<uri-variable name="foo" expression="headers.bar"/>
</outbound-gateway>
<beans:bean id="bodyExtractor"
class="org.springframework.integration.webflux.support.ClientHttpResponseBodyExtractor"/>
</beans:beans>

11
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -40,6 +40,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.function.BodyExtractor;
import org.springframework.web.reactive.function.client.WebClient;
/**
@@ -62,6 +63,9 @@ public class WebFluxOutboundGatewayParserTests {
@Autowired
private WebClient webClient;
@Autowired
private BodyExtractor<?, ?> bodyExtractor;
@Autowired
private ApplicationContext applicationContext;
@@ -82,7 +86,7 @@ public class WebFluxOutboundGatewayParserTests {
assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset"));
assertEquals(true, handlerAccessor.getPropertyValue("extractPayload"));
assertEquals(false, handlerAccessor.getPropertyValue("transferCookies"));
assertEquals(false, handlerAccessor.getPropertyValue("replyToFlux"));
assertEquals(false, handlerAccessor.getPropertyValue("replyPayloadToFlux"));
}
@Test
@@ -124,7 +128,8 @@ public class WebFluxOutboundGatewayParserTests {
assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2"));
assertEquals("responseHeader", mappedResponseHeaders[0]);
assertEquals(true, handlerAccessor.getPropertyValue("transferCookies"));
assertEquals(true, handlerAccessor.getPropertyValue("replyToFlux"));
assertEquals(true, handlerAccessor.getPropertyValue("replyPayloadToFlux"));
assertSame(this.bodyExtractor, handlerAccessor.getPropertyValue("bodyExtractor"));
}
}

56
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.webflux.outbound;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
@@ -33,11 +34,13 @@ import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.http.HttpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.webflux.support.ClientHttpResponseBodyExtractor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.support.ErrorMessage;
@@ -215,7 +218,7 @@ public class WebFluxRequestExecutingMessageHandlerTests {
QueueChannel replyChannel = new QueueChannel();
reactiveHandler.setOutputChannel(replyChannel);
reactiveHandler.setExpectedResponseType(String.class);
reactiveHandler.setReplyToFlux(true);
reactiveHandler.setReplyPayloadToFlux(true);
reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
@@ -232,4 +235,55 @@ public class WebFluxRequestExecutingMessageHandlerTests {
.verifyComplete();
}
@Test
public void testClientHttpResponseAsReply() {
ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
response.setStatusCode(HttpStatus.OK);
response.getHeaders().setContentType(MediaType.TEXT_PLAIN);
DataBufferFactory bufferFactory = response.bufferFactory();
Flux<DataBuffer> data =
Flux.just(bufferFactory.wrap("foo".getBytes()),
bufferFactory.wrap("bar".getBytes()),
bufferFactory.wrap("baz".getBytes()));
return response.writeWith(data)
.then(Mono.defer(response::setComplete));
});
WebClient webClient = WebClient.builder()
.clientConnector(httpConnector)
.build();
String destinationUri = "http://www.springsource.org/spring-integration";
WebFluxRequestExecutingMessageHandler reactiveHandler =
new WebFluxRequestExecutingMessageHandler(destinationUri, webClient);
QueueChannel replyChannel = new QueueChannel();
reactiveHandler.setOutputChannel(replyChannel);
reactiveHandler.setBodyExtractor(new ClientHttpResponseBodyExtractor());
reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
Message<?> receive = replyChannel.receive(10_000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(ClientHttpResponse.class));
ClientHttpResponse response = (ClientHttpResponse) receive.getPayload();
assertEquals(HttpStatus.OK, response.getStatusCode());
assertEquals(MediaType.TEXT_PLAIN, response.getHeaders().getContentType());
StepVerifier.create(
response.getBody()
.map(dataBuffer -> new String(dataBuffer.asByteBuffer().array())))
.expectNext("foo", "bar", "baz")
.verifyComplete();
}
}

48
src/reference/asciidoc/webflux.adoc
View File

@@ -100,9 +100,13 @@ Together with the `ReactiveChannel` as an `outputChannel`, the `Mono<ClientRespo
Otherwise, it is treated as an `async` mode and the `Mono` response is adapted to an `SettableListenableFuture` for an asynchronous reply from the `WebFluxRequestExecutingMessageHandler`.
The target payload of the output message depends on the `WebFluxRequestExecutingMessageHandler` configuration.
The `setExpectedResponseType(Class<?>)` or `setExpectedResponseTypeExpression(Expression)` identifies the target type of the response body element conversion.
If `replyToFlux` is set to `true`, the response body is converted to a `Flux` with the provided `expectedResponseType` for each element and this `Flux` is sent as the payload downstream.
If the `replyPayloadToFlux` is set to `true`, the response body is converted to a `Flux` with the provided `expectedResponseType` for each element and this `Flux` is sent as the payload downstream.
A <<splitter,splitter>> afterwards can be used to iterate over this `Flux` in a reactive manner.
In addition a `BodyExtractor<?, ClientHttpResponse>` can be injected into the `WebFluxRequestExecutingMessageHandler` instead of `expectedResponseType` and `replyPayloadToFlux` properties.
It can be used for low-level access to the `ClientHttpResponse` and more control over body and HTTP headers conversion.
The `ClientHttpResponseBodyExtractor` is provided out-of-the-box as identity function to produce downstream the whole `ClientHttpResponse` and any other possible custom logic.
Also see <<http-outbound>> for more possible configuration options.
[[webflux-namespace]]
@@ -133,6 +137,48 @@ To include it in your configuration, simply provide the following namespace decl
==== Inbound
To configure Spring Integration WebFlux via XML you may use appropriate components from the mentioned `int-webflux` namespace - `inbound-channel-adapter` or `inbound-gateway` according request/response requirements respectively:
[source,xml]
----
<inbound-channel-adapter id="reactiveFullConfig" channel="requests"
path="test1"
auto-startup="false"
phase="101"
request-payload-type="byte[]"
error-channel="errorChannel"
payload-expression="payload"
supported-methods="PUT"
status-code-expression="'202'"
header-mapper="headerMapper"
codec-configurer="codecConfigurer"
reactive-adapter-registry="reactiveAdapterRegistry"
requested-content-type-resolver="requestedContentTypeResolver">
<request-mapping headers="foo"/>
<cross-origin origin="foo"
method="PUT"/>
<header name="foo" expression="'foo'"/>
</inbound-channel-adapter>
<inbound-gateway id="reactiveFullConfig" request-channel="requests"
path="test1"
auto-startup="false"
phase="101"
request-payload-type="byte[]"
error-channel="errorChannel"
payload-expression="payload"
supported-methods="PUT"
reply-timeout-status-code-expression="'504'"
header-mapper="headerMapper"
codec-configurer="codecConfigurer"
reactive-adapter-registry="reactiveAdapterRegistry"
requested-content-type-resolver="requestedContentTypeResolver">
<request-mapping headers="foo"/>
<cross-origin origin="foo"
method="PUT"/>
<header name="foo" expression="'foo'"/>
</inbound-gateway>
----
==== Outbound