WebFlux improvements

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

Rename `replyToFlux` property to the `replyPayloadToFlux`

Doc Polishing

Author: artembilan

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxInboundEndpointParser.java

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,6 +28,8 @@
 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());
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,7 +47,8 @@ protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parser
 					.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;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,11 +20,16 @@
 
 import 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;

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

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@
 import org.springframework.http.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 WebFluxRequestExecutingMessageHandler(Expression uriExpression, WebClient
 	 * 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 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 setReplyToFlux(boolean replyToFlux) {
-		this.replyToFlux = replyToFlux;
+	public void setBodyExtractor(BodyExtractor<?, ClientHttpResponse> bodyExtractor) {
+		this.bodyExtractor = bodyExtractor;
 	}
 
 	@Override
@@ -191,7 +206,7 @@ protected Object exchange(Supplier<URI> uriSupplier, HttpMethod httpMethod, Http
 								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 @@ protected Object exchange(Supplier<URI> uriSupplier, HttpMethod httpMethod, Http
 										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();
 								}

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/support/ClientHttpResponseBodyExtractor.java

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