Support non-Object types for WebFlux requests

* Add the support for a `Publisher`, `Resource` and `MultiValueMap`
into the `WebFluxRequestExecutingMessageHandler`
* Along side with the `WebFluxRequestExecutingMessageHandler.setPublisherElementType`
and `WebFluxRequestExecutingMessageHandler.setPublisherElementTypeExpression`,
add XSD support for the `publisher-element-type(-expression)`, which
is used for the element type when request body is a `Publisher`
* Polishing for `AbstractHttpRequestExecutingMessageHandler`

* Fix Sonar smells for affected classes
* Remove used imports

Doc polishing

spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java

@@ -30,6 +30,9 @@ import java.util.function.Supplier;
 
 import javax.xml.transform.Source;
 
+import org.reactivestreams.Publisher;
+
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
@@ -48,6 +51,7 @@ import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
 import org.springframework.integration.mapping.HeaderMapper;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.MessageBuilderFactory;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.util.Assert;
@@ -74,7 +78,7 @@ import org.springframework.web.util.UriComponentsBuilder;
  */
 public abstract class AbstractHttpRequestExecutingMessageHandler extends AbstractReplyProducingMessageHandler {
 
-	private static final List<HttpMethod> noBodyHttpMethods =
+	private static final List<HttpMethod> NO_BODY_HTTP_METHODS =
 			Arrays.asList(HttpMethod.GET, HttpMethod.HEAD, HttpMethod.TRACE);
 
 	private final Map<String, Expression> uriVariableExpressions = new HashMap<>();
@@ -140,7 +144,7 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 	 */
 	public void setHttpMethod(HttpMethod httpMethod) {
 		Assert.notNull(httpMethod, "'httpMethod' must not be null");
-		this.httpMethodExpression = new ValueExpression<HttpMethod>(httpMethod);
+		this.httpMethodExpression = new ValueExpression<>(httpMethod);
 	}
 
 	/**
@@ -193,7 +197,7 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 	 */
 	public void setExpectedResponseType(Class<?> expectedResponseType) {
 		Assert.notNull(expectedResponseType, "'expectedResponseType' must not be null");
-		this.expectedResponseTypeExpression = new ValueExpression<Class<?>>(expectedResponseType);
+		setExpectedResponseTypeExpression(new ValueExpression<>(expectedResponseType));
 	}
 
 	/**
@@ -261,20 +265,18 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 
 	@Override
 	protected void doInit() {
-		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
-		this.simpleEvaluationContext = ExpressionUtils.createSimpleEvaluationContext(this.getBeanFactory());
+		BeanFactory beanFactory = getBeanFactory();
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(beanFactory);
+		this.simpleEvaluationContext = ExpressionUtils.createSimpleEvaluationContext(beanFactory);
 	}
 
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		HttpMethod httpMethod = determineHttpMethod(requestMessage);
-
-		if (!shouldIncludeRequestBody(httpMethod) && this.extractPayloadExplicitlySet) {
-			if (logger.isWarnEnabled()) {
-				logger.warn("The 'extractPayload' attribute has no relevance for the current request " +
-						"since the HTTP Method is '" + httpMethod +
-						"', and no request body will be sent for that method.");
-			}
+		if (this.extractPayloadExplicitlySet && logger.isWarnEnabled() && !shouldIncludeRequestBody(httpMethod)) {
+			logger.warn("The 'extractPayload' attribute has no relevance for the current request " +
+					"since the HTTP Method is '" + httpMethod +
+					"', and no request body will be sent for that method.");
 		}
 
 		Object expectedResponseType = determineExpectedResponseType(requestMessage);
@@ -293,9 +295,10 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 				"'uriExpression' evaluation must result in a 'String' or 'URI' instance, not: "
 						+ (uri == null ? "null" : uri.getClass()));
 		Map<String, ?> uriVariables = determineUriVariables(requestMessage);
-		UriComponentsBuilder uriComponentsBuilder = uri instanceof String
-				? UriComponentsBuilder.fromUriString((String) uri)
-				: UriComponentsBuilder.fromUri((URI) uri);
+		UriComponentsBuilder uriComponentsBuilder =
+				uri instanceof String
+						? UriComponentsBuilder.fromUriString((String) uri)
+						: UriComponentsBuilder.fromUri((URI) uri);
 		UriComponents uriComponents = uriComponentsBuilder.buildAndExpand(uriVariables);
 		try {
 			return this.encodeUri ? uriComponents.toUri() : new URI(uriComponents.toUriString());
@@ -310,7 +313,7 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 			HttpHeaders httpHeaders = httpResponse.getHeaders();
 			Map<String, Object> headers = this.headerMapper.toHeaders(httpHeaders);
 			if (this.transferCookies) {
-				this.doConvertSetCookie(headers);
+				doConvertSetCookie(headers);
 			}
 
 			AbstractIntegrationMessageBuilder<?> replyBuilder = null;
@@ -355,8 +358,9 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 
 	private HttpEntity<?> generateHttpRequest(Message<?> message, HttpMethod httpMethod) {
 		Assert.notNull(message, "message must not be null");
-		return (this.extractPayload) ? this.createHttpEntityFromPayload(message, httpMethod)
-				: this.createHttpEntityFromMessage(message, httpMethod);
+		return this.extractPayload
+				? createHttpEntityFromPayload(message, httpMethod)
+				: createHttpEntityFromMessage(message, httpMethod);
 	}
 
 	private HttpEntity<?> createHttpEntityFromPayload(Message<?> message, HttpMethod httpMethod) {
@@ -371,16 +375,20 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 		}
 		// otherwise, we are creating a request with a body and need to deal with the content-type header as well
 		if (httpHeaders.getContentType() == null) {
-			MediaType contentType = (payload instanceof String)
-					? resolveContentType((String) payload, this.charset)
-					: resolveContentType(payload);
+			MediaType contentType =
+					payload instanceof String
+							? new MediaType("text", "plain", this.charset)
+							: resolveContentType(payload);
 			httpHeaders.setContentType(contentType);
 		}
-		if (MediaType.APPLICATION_FORM_URLENCODED.equals(httpHeaders.getContentType()) ||
-				MediaType.MULTIPART_FORM_DATA.equals(httpHeaders.getContentType())) {
-			if (!(payload instanceof MultiValueMap)) {
-				payload = this.convertToMultiValueMap((Map<?, ?>) payload);
-			}
+		if ((MediaType.APPLICATION_FORM_URLENCODED.equals(httpHeaders.getContentType()) ||
+				MediaType.MULTIPART_FORM_DATA.equals(httpHeaders.getContentType()))
+				&& !(payload instanceof MultiValueMap)) {
+
+			Assert.isInstanceOf(Map.class, payload,
+					() -> "For " + MediaType.APPLICATION_FORM_URLENCODED + " and " +
+							MediaType.MULTIPART_FORM_DATA + " media types the payload must be an instance of a Map.");
+			payload = convertToMultiValueMap((Map<?, ?>) payload);
 		}
 		return new HttpEntity<>(payload, httpHeaders);
 	}
@@ -409,34 +417,28 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 		else if (content instanceof Source) {
 			contentType = MediaType.TEXT_XML;
 		}
-		else if (content instanceof Map) {
+		else if (content instanceof Map && isFormData((Map<Object, ?>) content)) {
 			// We need to check separately for MULTIPART as well as URLENCODED simply because
 			// MultiValueMap<Object, Object> is actually valid content for serialization
-			if (this.isFormData((Map<Object, ?>) content)) {
-				if (this.isMultipart((Map<String, ?>) content)) {
-					contentType = MediaType.MULTIPART_FORM_DATA;
-				}
-				else {
-					contentType = MediaType.APPLICATION_FORM_URLENCODED;
-				}
+			if (isMultipart((Map<String, ?>) content)) {
+				contentType = MediaType.MULTIPART_FORM_DATA;
+			}
+			else {
+				contentType = MediaType.APPLICATION_FORM_URLENCODED;
 			}
 		}
-		if (contentType == null) {
+		if (contentType == null && !(content instanceof Publisher<?>)) {
 			contentType = new MediaType("application", "x-java-serialized-object");
 		}
 		return contentType;
 	}
 
 	private boolean shouldIncludeRequestBody(HttpMethod httpMethod) {
-		return !(CollectionUtils.containsInstance(noBodyHttpMethods, httpMethod));
-	}
-
-	private MediaType resolveContentType(String content, Charset charset) {
-		return new MediaType("text", "plain", charset);
+		return !(CollectionUtils.containsInstance(NO_BODY_HTTP_METHODS, httpMethod));
 	}
 
 	private MultiValueMap<Object, Object> convertToMultiValueMap(Map<?, ?> simpleMap) {
-		LinkedMultiValueMap<Object, Object> multipartValueMap = new LinkedMultiValueMap<Object, Object>();
+		LinkedMultiValueMap<Object, Object> multipartValueMap = new LinkedMultiValueMap<>();
 		for (Entry<?, ?> entry : simpleMap.entrySet()) {
 			Object key = entry.getKey();
 			Object value = entry.getValue();
@@ -444,7 +446,7 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 				value = Arrays.asList((Object[]) value);
 			}
 			if (value instanceof Collection) {
-				multipartValueMap.put(key, new ArrayList<Object>((Collection<?>) value));
+				multipartValueMap.put(key, new ArrayList<>((Collection<?>) value));
 			}
 			else {
 				multipartValueMap.add(key, value);
@@ -453,6 +455,18 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 		return multipartValueMap;
 	}
 
+	/**
+	 * If all keys and values are Strings, we'll consider the Map to be form data.
+	 */
+	private boolean isFormData(Map<Object, ?> map) {
+		for (Object key : map.keySet()) {
+			if (!(key instanceof String)) {
+				return false;
+			}
+		}
+		return true;
+	}
+
 	/**
 	 * If all keys are Strings, and some values are not Strings we'll consider
 	 * the Map to be multipart/form-data
@@ -479,21 +493,9 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 		return false;
 	}
 
-	/**
-	 * If all keys and values are Strings, we'll consider the Map to be form data.
-	 */
-	private boolean isFormData(Map<Object, ?> map) {
-		for (Object key : map.keySet()) {
-			if (!(key instanceof String)) {
-				return false;
-			}
-		}
-		return true;
-	}
-
 	private HttpMethod determineHttpMethod(Message<?> requestMessage) {
 		Object httpMethod = this.httpMethodExpression.getValue(this.evaluationContext, requestMessage);
-		Assert.state(httpMethod != null && (httpMethod instanceof String || httpMethod instanceof HttpMethod),
+		Assert.state((httpMethod instanceof String || httpMethod instanceof HttpMethod), () ->
 				"'httpMethodExpression' evaluation must result in an 'HttpMethod' enum or its String representation, " +
 						"not: " + (httpMethod == null ? "null" : httpMethod.getClass()));
 		if (httpMethod instanceof HttpMethod) {
@@ -503,36 +505,43 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac
 			try {
 				return HttpMethod.valueOf((String) httpMethod);
 			}
-			catch (Exception e) {
+			catch (Exception ex) {
 				throw new IllegalStateException("The 'httpMethodExpression' returned an invalid HTTP Method value: "
-						+ httpMethod);
+						+ httpMethod, ex);
 			}
 		}
 	}
 
 	private Object determineExpectedResponseType(Message<?> requestMessage) {
-		Object expectedResponseType = null;
-		if (this.expectedResponseTypeExpression != null) {
-			expectedResponseType = this.expectedResponseTypeExpression.getValue(this.evaluationContext, requestMessage);
+		return evaluateTypeFromExpression(requestMessage, this.expectedResponseTypeExpression, "expectedResponseType");
+	}
+
+	@Nullable
+	protected Object evaluateTypeFromExpression(Message<?> requestMessage, @Nullable Expression expression,
+			String property) {
+
+		Object type = null;
+		if (expression != null) {
+			type = expression.getValue(this.evaluationContext, requestMessage);
 		}
-		if (expectedResponseType != null) {
-			Assert.state(expectedResponseType instanceof Class<?>
-							|| expectedResponseType instanceof String
-							|| expectedResponseType instanceof ParameterizedTypeReference,
-					"'expectedResponseType' can be an instance of 'Class<?>', 'String' " +
+		if (type != null) {
+			Class<?> typeClass = type.getClass();
+			Assert.state(type instanceof Class<?>
+							|| type instanceof String
+							|| type instanceof ParameterizedTypeReference,
+					() -> "The '" + property + "' can be an instance of 'Class<?>', 'String' " +
 							"or 'ParameterizedTypeReference<?>'; " +
-							"evaluation resulted in a" + expectedResponseType.getClass() + ".");
-			if (expectedResponseType instanceof String && StringUtils.hasText((String) expectedResponseType)) {
+							"evaluation resulted in a " + typeClass + ".");
+			if (type instanceof String && StringUtils.hasText((String) type)) {
 				try {
-					expectedResponseType = ClassUtils.forName((String) expectedResponseType,
-							getApplicationContext().getClassLoader());
+					type = ClassUtils.forName((String) type, getApplicationContext().getClassLoader());
 				}
 				catch (ClassNotFoundException e) {
-					throw new IllegalStateException("Cannot load class for name: " + expectedResponseType, e);
+					throw new IllegalStateException("Cannot load class for name: " + type, e);
 				}
 			}
 		}
-		return expectedResponseType;
+		return type;
 	}
 
 	@SuppressWarnings("unchecked")

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

@@ -21,6 +21,7 @@ import org.w3c.dom.Element;
 import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.http.config.HttpOutboundChannelAdapterParser;
 import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler;
 import org.springframework.util.StringUtils;
@@ -36,6 +37,12 @@ public class WebFluxOutboundChannelAdapterParser extends HttpOutboundChannelAdap
 
 	@Override
 	protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parserContext) {
+		return buildWebFluxRequestExecutingMessageHandler(element, parserContext);
+	}
+
+	static BeanDefinitionBuilder buildWebFluxRequestExecutingMessageHandler(Element element,
+			ParserContext parserContext) {
+
 		BeanDefinitionBuilder builder =
 				BeanDefinitionBuilder.genericBeanDefinition(WebFluxRequestExecutingMessageHandler.class);
 
@@ -46,6 +53,28 @@ public class WebFluxOutboundChannelAdapterParser extends HttpOutboundChannelAdap
 					.addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef));
 		}
 
+		String type = element.getAttribute("publisher-element-type");
+		String typeExpression = element.getAttribute("publisher-element-type-expression");
+
+		boolean hasType = StringUtils.hasText(type);
+		boolean hasTypeExpression = StringUtils.hasText(typeExpression);
+
+		if (hasType && hasTypeExpression) {
+			parserContext.getReaderContext()
+					.error("The 'publisher-element-type' and 'publisher-element-type-expression' " +
+							"are mutually exclusive. You can only have one or the other", element);
+		}
+
+		if (hasType) {
+			builder.addPropertyValue("publisherElementType", type);
+		}
+		else if (hasTypeExpression) {
+			builder.addPropertyValue("publisherElementTypeExpression",
+					BeanDefinitionBuilder.rootBeanDefinition(ExpressionFactoryBean.class)
+							.addConstructorArgValue(typeExpression)
+							.getBeanDefinition());
+		}
+
 		return builder;
 	}
 

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

@@ -18,13 +18,10 @@ package org.springframework.integration.webflux.config;
 
 import org.w3c.dom.Element;
 
-import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.http.config.HttpOutboundGatewayParser;
-import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler;
-import org.springframework.util.StringUtils;
 
 /**
  * Parser for the 'outbound-gateway' element of the webflux namespace.
@@ -38,15 +35,7 @@ public class WebFluxOutboundGatewayParser extends HttpOutboundGatewayParser {
 	@Override
 	protected BeanDefinitionBuilder getBuilder(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder =
-				BeanDefinitionBuilder.genericBeanDefinition(WebFluxRequestExecutingMessageHandler.class);
-
-		String webClientRef = element.getAttribute("web-client");
-		if (StringUtils.hasText(webClientRef)) {
-			builder.getBeanDefinition()
-					.getConstructorArgumentValues()
-					.addIndexedArgumentValue(1, new RuntimeBeanReference(webClientRef));
-		}
-
+				WebFluxOutboundChannelAdapterParser.buildWebFluxRequestExecutingMessageHandler(element, parserContext);
 		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/outbound/WebFluxRequestExecutingMessageHandler.java

@@ -20,7 +20,10 @@ import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.util.function.Supplier;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.expression.Expression;
@@ -28,16 +31,21 @@ import org.springframework.expression.common.LiteralExpression;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
 import org.springframework.http.ReactiveHttpInputMessage;
 import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.reactive.ClientHttpRequest;
 import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.http.outbound.AbstractHttpRequestExecutingMessageHandler;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
+import org.springframework.util.MultiValueMap;
 import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.BodyExtractors;
+import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.BodyInserters;
 import org.springframework.web.reactive.function.client.ClientResponse;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -57,6 +65,7 @@ import reactor.core.publisher.Mono;
  * @since 5.0
  *
  * @see org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler
+ * @see WebClient
  */
 public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestExecutingMessageHandler {
 
@@ -66,6 +75,8 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
 
 	private BodyExtractor<?, ClientHttpResponse> bodyExtractor;
 
+	private Expression publisherElementTypeExpression;
+
 	/**
 	 * Create a handler that will send requests to the provided URI.
 	 * @param uri The URI.
@@ -143,6 +154,31 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
 		this.bodyExtractor = bodyExtractor;
 	}
 
+	/**
+	 * Configure a type for a request {@link Publisher} elements.
+	 * @param publisherElementType the type of the request {@link Publisher} elements.
+	 * @since 5.2
+	 * @see BodyInserters#fromPublisher(Publisher, Class)
+	 */
+	public void setPublisherElementType(Class<?> publisherElementType) {
+		Assert.notNull(publisherElementType, "'publisherElementType' must not be null");
+		setPublisherElementTypeExpression(new ValueExpression<>(publisherElementType));
+
+	}
+
+	/**
+	 * Configure a SpEL expression to evaluate a request {@link Publisher} elements type at runtime against
+	 * a request message.
+	 * @param publisherElementTypeExpression the expression to evaluate a type for the request
+	 * {@link Publisher} elements.
+	 * @since 5.2
+	 * @see BodyInserters#fromPublisher(Publisher, Class)
+	 * @see BodyInserters#fromPublisher(Publisher, ParameterizedTypeReference)
+	 */
+	public void setPublisherElementTypeExpression(Expression publisherElementTypeExpression) {
+		this.publisherElementTypeExpression = publisherElementTypeExpression;
+	}
+
 	@Override
 	public String getComponentType() {
 		return (isExpectReply() ? "webflux:outbound-gateway" : "webflux:outbound-channel-adapter");
@@ -156,9 +192,9 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
 				this.webClient.method(httpMethod)
 						.uri(b -> uriSupplier.get())
 						.headers(headers -> headers.putAll(httpRequest.getHeaders()));
-
-		if (httpRequest.hasBody()) {
-			requestSpec.body(BodyInserters.fromObject(httpRequest.getBody())); // NOSONAR protected with hasBody()
+		BodyInserter<?, ? super ClientHttpRequest> inserter = buildBodyInserterForRequest(requestMessage, httpRequest);
+		if (inserter != null) {
+			requestSpec.body(inserter);
 		}
 
 		Mono<ClientResponse> responseMono =
@@ -256,4 +292,66 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx
 		}
 	}
 
+	@Nullable
+	private BodyInserter<?, ? super ClientHttpRequest> buildBodyInserterForRequest(Message<?> requestMessage,
+			HttpEntity<?> httpRequest) {
+
+		Object requestBody = httpRequest.getBody();
+		if (requestBody == null) {
+			return null;
+		}
+
+		BodyInserter<?, ? super ClientHttpRequest> inserter = null;
+		if (requestBody instanceof Resource) {
+			inserter = BodyInserters.fromResource((Resource) requestBody);
+		}
+		else if (requestBody instanceof Publisher<?>) {
+			inserter = buildBodyInserterForPublisher(requestMessage, (Publisher<?>) requestBody);
+		}
+		else if (requestBody instanceof MultiValueMap<?, ?>) {
+			inserter = buildBodyInserterForMultiValueMap((MultiValueMap<?, ?>) requestBody,
+					httpRequest.getHeaders().getContentType());
+		}
+
+		if (inserter == null) {
+			inserter = BodyInserters.fromObject(requestBody);
+		}
+		return inserter;
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T, P extends Publisher<T>> BodyInserter<P, ? super ClientHttpRequest> buildBodyInserterForPublisher(
+			Message<?> requestMessage, P publisher) {
+
+		BodyInserter<P, ? super ClientHttpRequest> inserter;
+		Object publisherElementType = evaluateTypeFromExpression(requestMessage,
+				this.publisherElementTypeExpression, "publisherElementType");
+		if (publisherElementType instanceof Class<?>) {
+			inserter = BodyInserters.fromPublisher(publisher, (Class<T>) publisherElementType);
+		}
+		else if (publisherElementType instanceof ParameterizedTypeReference<?>) {
+			inserter = BodyInserters.fromPublisher(publisher, (ParameterizedTypeReference<T>) publisherElementType);
+		}
+		else {
+			inserter = BodyInserters.fromPublisher(publisher, (Class<T>) Object.class);
+		}
+		return inserter;
+	}
+
+	@Nullable
+	@SuppressWarnings("unchecked")
+	private BodyInserters.FormInserter<?> buildBodyInserterForMultiValueMap(
+			MultiValueMap<?, ?> requestBody, MediaType contentType) {
+
+		if (MediaType.APPLICATION_FORM_URLENCODED.equals(contentType)) {
+			return BodyInserters.fromFormData((MultiValueMap<String, String>) requestBody);
+		}
+		else if (MediaType.MULTIPART_FORM_DATA.equals(contentType)) {
+			return BodyInserters.fromMultipartData((MultiValueMap<String, ?>) requestBody);
+		}
+		else {
+			return null;
+		}
+	}
+
 }

spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.2.xsd

@@ -376,6 +376,29 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="publisher-element-type" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						The type for a request 'Publisher' elements.
+						This attribute cannot be provided if the expected-response-type-expression has a value
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation>
+							<tool:expected-type type="java.lang.Class" />
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="publisher-element-type-expression" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						SpEL expression to determine the type for a request 'Publisher' elements against a request
+						message. The returned value of the expression could be an instance of java.lang.Class,
+						or java.lang.String representing a fully qualified class name, or 'ParameterizedTypeReference'.
+						This attribute cannot be provided if 'publisher-element-type' has a value
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 
@@ -513,6 +536,29 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="publisher-element-type" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								The type for a request 'Publisher' elements.
+								This attribute cannot be provided if the expected-response-type-expression has a value
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation>
+									<tool:expected-type type="java.lang.Class" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="publisher-element-type-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								SpEL expression to determine the type for a request 'Publisher' elements against a request
+								message. The returned value of the expression could be an instance of java.lang.Class,
+								or java.lang.String representing a fully qualified class name, or 'ParameterizedTypeReference'.
+								This attribute cannot be provided if 'publisher-element-type' has a value
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests-context.xml

@@ -13,7 +13,8 @@
 	<outbound-channel-adapter id="reactiveMinimalConfig" url="http://localhost/test1" channel="requests"/>
 
 	<outbound-channel-adapter id="reactiveWebClientConfig" url="http://localhost/test1" channel="requests"
-									   web-client="webClient"/>
+							  web-client="webClient"
+							  publisher-element-type="java.util.Date"/>
 
 	<beans:bean id="webClient" class="org.springframework.web.reactive.function.client.WebClient"
 				factory-method="create"/>

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests.java

@@ -19,6 +19,7 @@ package org.springframework.integration.webflux.config;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.Charset;
+import java.util.Date;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -82,6 +83,9 @@ public class WebFluxOutboundChannelAdapterParserTests {
 	public void reactiveWebClientConfig() {
 		assertThat(TestUtils.getPropertyValue(this.reactiveWebClientConfig, "handler.webClient"))
 				.isSameAs(this.webClient);
+		assertThat(TestUtils.getPropertyValue(this.reactiveWebClientConfig,
+				"handler.publisherElementTypeExpression.value"))
+				.isSameAs(Date.class);
 	}
 
 }

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

@@ -37,7 +37,8 @@
 					  auto-startup="false"
 					  transfer-cookies="true"
 					  reply-payload-to-flux="true"
-					  body-extractor="bodyExtractor">
+					  body-extractor="bodyExtractor"
+					  publisher-element-type-expression="headers.elementType">
 		<uri-variable name="foo" expression="headers.bar"/>
 	</outbound-gateway>
 

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

@@ -126,6 +126,8 @@ public class WebFluxOutboundGatewayParserTests {
 		assertThat(handlerAccessor.getPropertyValue("transferCookies")).isEqualTo(true);
 		assertThat(handlerAccessor.getPropertyValue("replyPayloadToFlux")).isEqualTo(true);
 		assertThat(handlerAccessor.getPropertyValue("bodyExtractor")).isSameAs(this.bodyExtractor);
+		assertThat(handlerAccessor.getPropertyValue("publisherElementTypeExpression.expression"))
+				.isEqualTo("headers.elementType");
 	}
 
 }

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

@@ -211,7 +211,7 @@ public class WebFluxRequestExecutingMessageHandlerTests {
 		reactiveHandler.setExpectedResponseType(String.class);
 		reactiveHandler.setReplyPayloadToFlux(true);
 
-		reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
+		reactiveHandler.handleMessage(MessageBuilder.withPayload(Mono.just("hello, world")).build());
 
 		Message<?> receive = replyChannel.receive(10_000);
 

src/reference/asciidoc/webflux.adoc

@@ -138,6 +138,11 @@ In addition a `BodyExtractor<?, ClientHttpResponse>` can be injected into the `W
 It can be used for low-level access to the `ClientHttpResponse` and more control over body and HTTP headers conversion.
 Spring Integration provides `ClientHttpResponseBodyExtractor` as a identity function to produce (downstream) the whole `ClientHttpResponse` and any other possible custom logic.
 
+Starting with version 5.2, the `WebFluxRequestExecutingMessageHandler` supports reactive `Publisher`, `Resource`, and `MultiValueMap` types as the request message payload.
+A respective `BodyInserter` is used internally to be populated into the `WebClient.RequestBodySpec`.
+When the payload is a reactive `Publisher`, a configured `publisherElementType` or `publisherElementTypeExpression` can be used to determine a type for the publisher's element type.
+The expression must be resolved to a `Class<?>`, `String` which is resolved to the target `Class<?>` or `ParameterizedTypeReference`.
+
 See <<http-outbound>> for more possible configuration options.
 
 [[webflux-namespace]]

src/reference/asciidoc/whats-new.adoc

@@ -60,3 +60,9 @@ See <<note-nio>> for more information.
 
 The `AbstractMailReceiver` has now an `autoCloseFolder` option (`true` by default), to disable an automatic folder close after a fetch, but populate `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header instead for downstream interaction.
 See <<mail-inbound>> for more information.
+
+[[x5.2-webflux]]
+==== WebFlux Changes
+
+The `WebFluxRequestExecutingMessageHandler` now supports a `Publisher`, `Resource` and `MultiValueMap` as a request message `payload`.
+See <<webflux>> for more information.