Add validation support into WebFlux Inbound (#2977)

* Add validation support into WebFlux Inbound

See https://stackoverflow.com/questions/56729517/spring-integration-webflux-validate-input-json-body

* Introduce a `WebFluxInboundEndpoint.validator` option to validate
a request payload after conversion HTTP request into a `Publisher`
* Expose a `WebFluxInboundEndpointSpec.validator()` option
* Introduce a `IntegrationWebExchangeBindException` to extend a
`WebExchangeBindException` for Spring Integration use-case.
such an exception is used in the `ResponseStatusExceptionHandler`
extensions, like one in Spring Boot for proper response rendering
* Test and document the solution

* * Remove unused imports from IntegrationWebExchangeBindException

spring-integration-http/src/main/java/org/springframework/integration/http/support/IntegrationWebExchangeBindException.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.http.support;
+
+import org.springframework.validation.BindingResult;
+import org.springframework.validation.ObjectError;
+import org.springframework.web.bind.support.WebExchangeBindException;
+
+/**
+ * A {@link WebExchangeBindException} extension for validation error with a failed
+ * message context.
+ * We can't rely on the default {@link WebExchangeBindException} behavior since
+ * there is no POJO method invocation in Spring Integration Web endpoint implementations.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.2
+ */
+@SuppressWarnings("serial")
+public class IntegrationWebExchangeBindException extends WebExchangeBindException {
+
+	private final String endpointId;
+
+	private final Object failedPayload;
+
+	public IntegrationWebExchangeBindException(String endpointId, Object failedPayload,
+			BindingResult bindingResult) {
+
+		super(null, bindingResult); // NOSONAR - we ignore a MethodParameter in favor of payload and endpoint context
+		this.endpointId = endpointId;
+		this.failedPayload = failedPayload;
+	}
+
+
+	@Override
+	public String getMessage() {
+		BindingResult bindingResult = getBindingResult();
+		StringBuilder sb = new StringBuilder("Validation failed for payload ")
+				.append(this.failedPayload)
+				.append(" in the endpoint ")
+				.append(this.endpointId)
+				.append(", with ")
+				.append(bindingResult.getErrorCount())
+				.append(" error(s): ");
+		for (ObjectError error : bindingResult.getAllErrors()) {
+			sb.append("[").append(error).append("] ");
+		}
+		return sb.toString();
+	}
+
+}

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

@@ -20,6 +20,7 @@ import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.http.codec.ServerCodecConfigurer;
 import org.springframework.integration.http.dsl.HttpInboundEndpointSupportSpec;
 import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
+import org.springframework.validation.Validator;
 import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
 
 /**
@@ -53,4 +54,15 @@ public class WebFluxInboundEndpointSpec
 		return this;
 	}
 
+	/**
+	 * Specify a {@link Validator} to validate a converted payload from request.
+	 * @param validator the {@link Validator} to use.
+	 * @return the spec
+	 * @since 5.2
+	 */
+	public WebFluxInboundEndpointSpec validator(Validator validator) {
+		this.target.setValidator(validator);
+		return this;
+	}
+
 }

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

@@ -48,15 +48,20 @@ import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.integration.expression.ExpressionEvalMap;
 import org.springframework.integration.http.inbound.BaseHttpInboundEndpoint;
+import org.springframework.integration.http.support.IntegrationWebExchangeBindException;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MultiValueMap;
+import org.springframework.validation.BeanPropertyBindingResult;
+import org.springframework.validation.ValidationUtils;
+import org.springframework.validation.Validator;
 import org.springframework.web.reactive.HandlerMapping;
 import org.springframework.web.reactive.accept.HeaderContentTypeResolver;
 import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
 import org.springframework.web.server.NotAcceptableStatusException;
+import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.server.ServerWebExchange;
 import org.springframework.web.server.UnsupportedMediaTypeStatusException;
 import org.springframework.web.server.WebHandler;
@@ -89,6 +94,8 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 
 	private ReactiveAdapterRegistry adapterRegistry = new ReactiveAdapterRegistry();
 
+	private Validator validator;
+
 	public WebFluxInboundEndpoint() {
 		this(true);
 	}
@@ -126,6 +133,15 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 		this.adapterRegistry = adapterRegistry;
 	}
 
+	/**
+	 * Specify a {@link Validator} to validate a converted payload from request.
+	 * @param validator the {@link Validator} to use.
+	 * @since 5.2
+	 */
+	public void setValidator(Validator validator) {
+		this.validator = validator;
+	}
+
 	@Override
 	public String getComponentType() {
 		return super.getComponentType().replaceFirst("http", "webflux");
@@ -138,7 +154,8 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 				return doHandle(exchange);
 			}
 			else {
-				return serviceUnavailableResponse(exchange);
+				return Mono.error(new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "Endpoint is stopped"))
+						.then();
 			}
 		});
 	}
@@ -146,8 +163,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 	private Mono<Void> doHandle(ServerWebExchange exchange) {
 		return extractRequestBody(exchange)
 				.doOnSubscribe(s -> this.activeCount.incrementAndGet())
-				.cast(Object.class)
-				.switchIfEmpty(Mono.just(exchange.getRequest().getQueryParams()))
 				.map(body ->
 						new RequestEntity<>(body, exchange.getRequest().getHeaders(),
 								exchange.getRequest().getMethod(), exchange.getRequest().getURI()))
@@ -168,10 +183,12 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 
 	private Mono<?> extractRequestBody(ServerWebExchange exchange) {
 		if (isReadable(exchange.getRequest().getMethod())) {
-			return extractReadableRequestBody(exchange);
+			return extractReadableRequestBody(exchange)
+					.cast(Object.class)
+					.switchIfEmpty(queryParams(exchange));
 		}
 		else {
-			return Mono.just(exchange.getRequest().getQueryParams());
+			return queryParams(exchange);
 		}
 	}
 
@@ -227,10 +244,16 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 		Map<String, Object> readHints = Collections.emptyMap();
 		if (adapter != null && adapter.isMultiValue()) {
 			Flux<?> flux = httpMessageReader.read(bodyType, elementType, request, response, readHints);
+			if (this.validator != null) {
+				flux = flux.doOnNext(this::validate);
+			}
 			return Mono.just(adapter.fromPublisher(flux));
 		}
 		else {
 			Mono<?> mono = httpMessageReader.readMono(bodyType, elementType, request, response, readHints);
+			if (this.validator != null) {
+				mono = mono.doOnNext(this::validate);
+			}
 			if (adapter != null) {
 				return Mono.just(adapter.fromPublisher(mono));
 			}
@@ -240,6 +263,14 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 		}
 	}
 
+	private void validate(Object value) {
+		BeanPropertyBindingResult errors = new BeanPropertyBindingResult(value, "requestPayload");
+		ValidationUtils.invokeValidator(this.validator, value, errors);
+		if (errors.hasErrors()) {
+			throw new IntegrationWebExchangeBindException(getComponentName(), value, errors);
+		}
+	}
+
 	private Mono<Tuple2<Message<Object>, RequestEntity<?>>> buildMessage(RequestEntity<?> httpEntity,
 			ServerWebExchange exchange) {
 
@@ -435,18 +466,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 		return Mono.error(new NotAcceptableStatusException(producibleMediaTypes));
 	}
 
-	private ResolvableType getElementType(ReactiveAdapter adapter, ResolvableType genericType) {
-		if (adapter.isNoValue()) {
-			return ResolvableType.forClass(Void.class);
-		}
-		else if (genericType != ResolvableType.NONE) {
-			return genericType;
-		}
-		else {
-			return ResolvableType.forClass(Object.class);
-		}
-	}
-
 	private List<MediaType> getProducibleMediaTypes(ResolvableType elementType) {
 		return this.codecConfigurer.getWriters()
 				.stream()
@@ -488,20 +507,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 		return (mediaTypes.isEmpty() ? Collections.singletonList(MediaType.ALL) : mediaTypes);
 	}
 
-	private List<MediaType> getProducibleTypes(ServerWebExchange exchange,
-			Supplier<List<MediaType>> producibleTypesSupplier) {
-
-		Set<MediaType> mediaTypes = exchange.getAttribute(HandlerMapping.PRODUCIBLE_MEDIA_TYPES_ATTRIBUTE);
-		return (mediaTypes != null ? new ArrayList<>(mediaTypes) : producibleTypesSupplier.get());
-	}
-
-	private MediaType selectMoreSpecificMediaType(MediaType acceptable, MediaType producible) {
-		MediaType producibleToUse = producible.copyQualityValue(acceptable);
-		Comparator<MediaType> comparator = MediaType.SPECIFICITY_COMPARATOR;
-		return (comparator.compare(acceptable, producibleToUse) <= 0 ? acceptable : producibleToUse);
-	}
-
-
 	private Mono<Void> setStatusCode(ServerWebExchange exchange, RequestEntity<?> requestEntity) {
 		ServerHttpResponse response = exchange.getResponse();
 		if (getStatusCodeExpression() != null) {
@@ -510,19 +515,36 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 				response.setStatusCode(httpStatus);
 			}
 		}
-
 		return response.setComplete();
 	}
 
-	private Mono<Void> serviceUnavailableResponse(ServerWebExchange exchange) {
+	private static ResolvableType getElementType(ReactiveAdapter adapter, ResolvableType genericType) {
-		if (logger.isDebugEnabled()) {
+		if (adapter.isNoValue()) {
-			logger.debug("Endpoint is stopped; returning status " + HttpStatus.SERVICE_UNAVAILABLE);
+			return ResolvableType.forClass(Void.class);
 		}
-		ServerHttpResponse response = exchange.getResponse();
+		else if (genericType != ResolvableType.NONE) {
-		response.setStatusCode(HttpStatus.SERVICE_UNAVAILABLE);
+			return genericType;
-		return response.writeWith(
+		}
-				Mono.just(response.bufferFactory()
+		else {
-						.wrap("Endpoint is stopped".getBytes())));
+			return ResolvableType.forClass(Object.class);
+		}
+	}
+
+	private static List<MediaType> getProducibleTypes(ServerWebExchange exchange,
+			Supplier<List<MediaType>> producibleTypesSupplier) {
+
+		Set<MediaType> mediaTypes = exchange.getAttribute(HandlerMapping.PRODUCIBLE_MEDIA_TYPES_ATTRIBUTE);
+		return (mediaTypes != null ? new ArrayList<>(mediaTypes) : producibleTypesSupplier.get());
+	}
+
+	private static Mono<?> queryParams(ServerWebExchange exchange) {
+		return Mono.just(exchange.getRequest().getQueryParams());
+	}
+
+	private static MediaType selectMoreSpecificMediaType(MediaType acceptable, MediaType producible) {
+		MediaType producibleToUse = producible.copyQualityValue(acceptable);
+		Comparator<MediaType> comparator = MediaType.SPECIFICITY_COMPARATOR;
+		return (comparator.compare(acceptable, producibleToUse) <= 0 ? acceptable : producibleToUse);
 	}
 
 }

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java

@@ -22,10 +22,12 @@ import static org.springframework.test.web.servlet.request.MockMvcRequestBuilder
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
 
 import java.security.Principal;
+import java.time.Duration;
 import java.util.Collections;
 
 import javax.annotation.Resource;
 
+import org.hamcrest.Matchers;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -81,8 +83,12 @@ import org.springframework.test.web.reactive.server.WebTestClient;
 import org.springframework.test.web.servlet.MockMvc;
 import org.springframework.test.web.servlet.setup.MockMvcBuilders;
 import org.springframework.util.MultiValueMap;
+import org.springframework.util.StringUtils;
+import org.springframework.validation.Errors;
+import org.springframework.validation.Validator;
 import org.springframework.web.context.WebApplicationContext;
 import org.springframework.web.reactive.config.EnableWebFlux;
+import org.springframework.web.reactive.config.WebFluxConfigurer;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.util.UriComponentsBuilder;
 
@@ -135,7 +141,7 @@ public class WebFluxDslTests {
 				WebTestClient.bindToApplicationContext(this.wac)
 						.apply(SecurityMockServerConfigurers.springSecurity())
 						.configureClient()
-//						.responseTimeout(Duration.ofSeconds(600))
+						.responseTimeout(Duration.ofSeconds(600))
 						.build();
 	}
 
@@ -237,7 +243,9 @@ public class WebFluxDslTests {
 				.headers(headers -> headers.setBasicAuth("guest", "guest"))
 				.body(Mono.just("foo\nbar\nbaz"), String.class)
 				.exchange()
-				.expectStatus().isEqualTo(HttpStatus.BAD_GATEWAY);
+				.expectStatus().isEqualTo(HttpStatus.BAD_GATEWAY)
+				.expectBody(String.class)
+				.value(Matchers.containsString("errorTest"));
 	}
 
 	@Test
@@ -256,7 +264,7 @@ public class WebFluxDslTests {
 	}
 
 	@Test
-	public void testDynamicHttpEndpoint() throws Exception {
+	public void testDynamicHttpEndpoint() {
 		IntegrationFlow flow =
 				IntegrationFlows.from(WebFlux.inboundGateway("/dynamic")
 						.requestMapping(r -> r.params("name"))
@@ -282,12 +290,48 @@ public class WebFluxDslTests {
 				.isNotFound();
 	}
 
+	@Autowired
+	private Validator validator;
+
+	@Test
+	public void testValidation() {
+		IntegrationFlow flow =
+				IntegrationFlows.from(
+						WebFlux.inboundGateway("/validation")
+								.requestMapping((mapping) -> mapping
+										.methods(HttpMethod.POST)
+										.consumes(MediaType.APPLICATION_JSON_VALUE))
+								.requestPayloadType(
+										ResolvableType.forClassWithGenerics(Flux.class, TestModel.class))
+								.validator(this.validator))
+						.bridge()
+						.get();
+
+		IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
+				this.integrationFlowContext.registration(flow).register();
+
+		this.webTestClient.post().uri("/validation")
+				.headers(headers -> headers.setBasicAuth("guest", "guest"))
+				.contentType(MediaType.APPLICATION_JSON)
+				.syncBody("{\"name\": \"\"}")
+				.exchange()
+				.expectStatus().isBadRequest();
+
+		flowRegistration.destroy();
+	}
+
+
 	@Configuration
 	@EnableWebFlux
 	@EnableWebSecurity
 	@EnableWebFluxSecurity
 	@EnableIntegration
-	public static class ContextConfiguration extends WebSecurityConfigurerAdapter {
+	public static class ContextConfiguration extends WebSecurityConfigurerAdapter implements WebFluxConfigurer {
+
+		@Override
+		public Validator getValidator() {
+			return webFluxValidator();
+		}
 
 		@Bean
 		public UserDetails userDetails() {
@@ -411,6 +455,42 @@ public class WebFluxDslTests {
 			return new AffirmativeBased(Collections.singletonList(new RoleVoter()));
 		}
 
+		@Bean
+		public Validator webFluxValidator() {
+			return new TestModelValidator();
+		}
+
+	}
+
+	public static class TestModel {
+
+		private String name;
+
+		public String getName() {
+			return this.name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+	}
+
+	private static class TestModelValidator implements Validator {
+
+		@Override
+		public boolean supports(Class<?> clazz) {
+			return TestModel.class.isAssignableFrom(clazz);
+		}
+
+		@Override
+		public void validate(Object target, Errors errors) {
+			TestModel testModel = (TestModel) target;
+			if (!StringUtils.hasText(testModel.getName())) {
+				errors.rejectValue("name", "Must not be empty");
+			}
+		}
+
 	}
 
 }

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java

@@ -69,8 +69,7 @@ public class WebFluxInboundEndpointTests {
 
 		this.webTestClient.get().uri("/test")
 				.exchange()
-				.expectStatus().isEqualTo(HttpStatus.SERVICE_UNAVAILABLE)
+				.expectStatus().isEqualTo(HttpStatus.SERVICE_UNAVAILABLE);
-				.expectBody(String.class).isEqualTo("Endpoint is stopped");
 	}
 
 	@Test

src/reference/asciidoc/webflux.adoc

@@ -91,6 +91,11 @@ See <<./http.adoc#http-request-mapping,Request Mapping Support>> and <<./http.ad
 
 When the request body is empty or `payloadExpression` returns `null`, the request params (`MultiValueMap<String, String>`) is used for a `payload` of the target message to process.
 
+Starting with version 5.2, the `WebFluxInboundEndpoint` can be configured with a `Validator`.
+It is used to validate elements in the `Publisher` to which a request has been converted by the `HttpMessageReader`.
+An invalid payload is rejected with an `IntegrationWebExchangeBindException` (a `WebExchangeBindException` extension), containing all the validation `Errors`.
+See more in Spring Framework https://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation[Reference Manual] about validation.
+
 [[webflux-outbound]]
 === WebFlux Outbound Components
 

src/reference/asciidoc/whats-new.adoc

@@ -97,7 +97,9 @@ See <<./mail.adoc#mail-inbound,Mail-receiving Channel Adapter>> for more informa
 ==== WebFlux Changes
 
 The `WebFluxRequestExecutingMessageHandler` now supports a `Publisher`, `Resource` and `MultiValueMap` as a request message `payload`.
+The `WebFluxInboundEndpoint` now supports a request payload validation.
 See <<./webflux.adoc#webflux,WebFlux Support>> for more information.
+
 [[x5.2-mongodb]]
 ==== MongoDb Changes