Wrap DecodingException in DefaultServerRequest

Prior to this commit, Spring WebFlux function would let
`DecodingException` thrown by codecs bubble up to the web handler level.
Since this exception is not handled by default there, the response would
be turned into a HTTP 500 status.

In the annotation model, `ArgumentResolver` implementations wrap this
exception with a `ServerWebInputException`, which itself extends
`ResponseStatusException`. The latter is supported by the error handling
infrastructure as a HTTP 400 response.

This commit ensures that `DecodingException` instances are properly
wrapped in `ServerWebInputException` at the `ServerRequest` level
directly, thus supporting all setup modes ("standalone" and through the
`DispatcherHandler`).

Fixes #22290

spring-webflux/src/main/java/org/springframework/web/reactive/function/server/DefaultServerRequest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-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.
@@ -33,6 +33,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.codec.DecodingException;
 import org.springframework.core.codec.Hints;
 import org.springframework.http.HttpCookie;
 import org.springframework.http.HttpHeaders;
@@ -48,6 +49,7 @@ import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.BodyExtractors;
 import org.springframework.web.reactive.function.UnsupportedMediaTypeException;
 import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.ServerWebInputException;
 import org.springframework.web.server.UnsupportedMediaTypeStatusException;
 import org.springframework.web.server.WebSession;
 import org.springframework.web.util.UriBuilder;
@@ -67,6 +69,9 @@ class DefaultServerRequest implements ServerRequest {
 							ex.getContentType(), ex.getSupportedMediaTypes(), ex.getBodyType()) :
 					new UnsupportedMediaTypeStatusException(ex.getMessage()));
 
+	private static final Function<DecodingException, ServerWebInputException> DECODING_MAPPER =
+			ex -> new ServerWebInputException("Failed to read HTTP message", null, ex);
+
 
 	private final ServerWebExchange exchange;
 
@@ -154,25 +159,29 @@ class DefaultServerRequest implements ServerRequest {
 	@Override
 	public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
 		Mono<T> mono = body(BodyExtractors.toMono(elementClass));
-		return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+		return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER)
+				.onErrorMap(DecodingException.class, DECODING_MAPPER);
 	}
 
 	@Override
 	public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
 		Mono<T> mono = body(BodyExtractors.toMono(typeReference));
-		return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+		return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER)
+				.onErrorMap(DecodingException.class, DECODING_MAPPER);
 	}
 
 	@Override
 	public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
 		Flux<T> flux = body(BodyExtractors.toFlux(elementClass));
-		return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+		return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER)
+				.onErrorMap(DecodingException.class, DECODING_MAPPER);
 	}
 
 	@Override
 	public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
 		Flux<T> flux = body(BodyExtractors.toFlux(typeReference));
-		return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+		return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER)
+				.onErrorMap(DecodingException.class, DECODING_MAPPER);
 	}
 
 	@Override