diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/support/IntegrationWebExchangeBindException.java b/spring-integration-http/src/main/java/org/springframework/integration/http/support/IntegrationWebExchangeBindException.java new file mode 100644 index 0000000000..17106fb50b --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java index 9f017eeb01..73887f2769 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/dsl/WebFluxInboundEndpointSpec.java +++ b/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; + } + } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index 898527fdf4..0d926893c3 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/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 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 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, 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 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 getProducibleTypes(ServerWebExchange exchange, - Supplier> producibleTypesSupplier) { - - Set 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 comparator = MediaType.SPECIFICITY_COMPARATOR; - return (comparator.compare(acceptable, producibleToUse) <= 0 ? acceptable : producibleToUse); - } - - private Mono 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 serviceUnavailableResponse(ServerWebExchange exchange) { - if (logger.isDebugEnabled()) { - logger.debug("Endpoint is stopped; returning status " + HttpStatus.SERVICE_UNAVAILABLE); + private static ResolvableType getElementType(ReactiveAdapter adapter, ResolvableType genericType) { + if (adapter.isNoValue()) { + return ResolvableType.forClass(Void.class); } - ServerHttpResponse response = exchange.getResponse(); - response.setStatusCode(HttpStatus.SERVICE_UNAVAILABLE); - return response.writeWith( - Mono.just(response.bufferFactory() - .wrap("Endpoint is stopped".getBytes()))); + else if (genericType != ResolvableType.NONE) { + return genericType; + } + else { + return ResolvableType.forClass(Object.class); + } + } + + private static List getProducibleTypes(ServerWebExchange exchange, + Supplier> producibleTypesSupplier) { + + Set 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 comparator = MediaType.SPECIFICITY_COMPARATOR; + return (comparator.compare(acceptable, producibleToUse) <= 0 ? acceptable : producibleToUse); } } diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java index bfd2e10ba7..4b9f2989d8 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java +++ b/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"); + } + } + } } diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java index fd33187708..93da402b40 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java +++ b/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) - .expectBody(String.class).isEqualTo("Endpoint is stopped"); + .expectStatus().isEqualTo(HttpStatus.SERVICE_UNAVAILABLE); } @Test diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc index 4d756acbe6..031333314d 100644 --- a/src/reference/asciidoc/webflux.adoc +++ b/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`) 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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b4fedad4dc..f869153bcd 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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