diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java index 52ff08d0c6..09c787c30f 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java @@ -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 noBodyHttpMethods = + private static final List NO_BODY_HTTP_METHODS = Arrays.asList(HttpMethod.GET, HttpMethod.HEAD, HttpMethod.TRACE); private final Map 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); + 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>(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 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 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) content)) { // We need to check separately for MULTIPART as well as URLENCODED simply because // MultiValueMap is actually valid content for serialization - if (this.isFormData((Map) content)) { - if (this.isMultipart((Map) content)) { - contentType = MediaType.MULTIPART_FORM_DATA; - } - else { - contentType = MediaType.APPLICATION_FORM_URLENCODED; - } + if (isMultipart((Map) 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 convertToMultiValueMap(Map simpleMap) { - LinkedMultiValueMap multipartValueMap = new LinkedMultiValueMap(); + LinkedMultiValueMap 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((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 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 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") diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParser.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParser.java index 06e3d2bbe2..8b0905bf8d 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParser.java +++ b/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; } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java index 48378753cc..acc2254e34 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParser.java +++ b/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; diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java index a8a7e94eb9..2d362b391a 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java +++ b/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 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 inserter = buildBodyInserterForRequest(requestMessage, httpRequest); + if (inserter != null) { + requestSpec.body(inserter); } Mono responseMono = @@ -256,4 +292,66 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx } } + @Nullable + private BodyInserter buildBodyInserterForRequest(Message requestMessage, + HttpEntity httpRequest) { + + Object requestBody = httpRequest.getBody(); + if (requestBody == null) { + return null; + } + + BodyInserter 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 > BodyInserter buildBodyInserterForPublisher( + Message requestMessage, P publisher) { + + BodyInserter inserter; + Object publisherElementType = evaluateTypeFromExpression(requestMessage, + this.publisherElementTypeExpression, "publisherElementType"); + if (publisherElementType instanceof Class) { + inserter = BodyInserters.fromPublisher(publisher, (Class) publisherElementType); + } + else if (publisherElementType instanceof ParameterizedTypeReference) { + inserter = BodyInserters.fromPublisher(publisher, (ParameterizedTypeReference) publisherElementType); + } + else { + inserter = BodyInserters.fromPublisher(publisher, (Class) 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) requestBody); + } + else if (MediaType.MULTIPART_FORM_DATA.equals(contentType)) { + return BodyInserters.fromMultipartData((MultiValueMap) requestBody); + } + else { + return null; + } + } + } diff --git a/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.2.xsd b/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.2.xsd index c78f9a191f..d47bfaaaa5 100644 --- a/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.2.xsd +++ b/spring-integration-webflux/src/main/resources/org/springframework/integration/webflux/config/spring-integration-webflux-5.2.xsd @@ -376,6 +376,29 @@ + + + + The type for a request 'Publisher' elements. + This attribute cannot be provided if the expected-response-type-expression has a value + + + + + + + + + + + + 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 + + + @@ -513,6 +536,29 @@ + + + + The type for a request 'Publisher' elements. + This attribute cannot be provided if the expected-response-type-expression has a value + + + + + + + + + + + + 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 + + + diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests-context.xml index 1eb84e6b5a..d16a911a62 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests-context.xml @@ -13,7 +13,8 @@ + web-client="webClient" + publisher-element-type="java.util.Date"/> diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests.java index eb31c8f55f..e9d99ee459 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundChannelAdapterParserTests.java +++ b/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); } } diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml index b123454d9d..6f1a069d60 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml +++ b/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"> diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java index 120296e1ea..cee78adc06 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java +++ b/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"); } } diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java index a129f556ae..64c9b3bd32 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java +++ b/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); diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc index 0e9eaf0bb7..71fdb33c5a 100644 --- a/src/reference/asciidoc/webflux.adoc +++ b/src/reference/asciidoc/webflux.adoc @@ -138,6 +138,11 @@ In addition a `BodyExtractor` 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 <> for more possible configuration options. [[webflux-namespace]] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 8c605fef79..0dcb33cbcf 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -60,3 +60,9 @@ See <> 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 <> for more information. + +[[x5.2-webflux]] +==== WebFlux Changes + +The `WebFluxRequestExecutingMessageHandler` now supports a `Publisher`, `Resource` and `MultiValueMap` as a request message `payload`. +See <> for more information.