Map to UnsupportedMediaTypeStatusException

This commit maps the UnsupportedMediaTypeException, used by both client
and server, to a server-side UnsupportedMediaTypeStatusException in the
functional web framework.

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

@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
@@ -39,7 +40,9 @@ import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.util.Assert;
 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.UnsupportedMediaTypeStatusException;
 import org.springframework.web.server.WebSession;
 
 /**
@@ -48,6 +51,12 @@ import org.springframework.web.server.WebSession;
  */
 class DefaultServerRequest implements ServerRequest {
 
+	private static final Function<UnsupportedMediaTypeException, UnsupportedMediaTypeStatusException> ERROR_MAPPER =
+			ex -> ex.getContentType()
+					.map(contentType -> new UnsupportedMediaTypeStatusException(contentType,
+							ex.getSupportedMediaTypes()))
+					.orElseGet(() -> new UnsupportedMediaTypeStatusException(ex.getMessage()));
+
 	private final ServerWebExchange exchange;
 
 	private final Headers headers;
@@ -100,12 +109,14 @@ class DefaultServerRequest implements ServerRequest {
 
 	@Override
 	public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
-		return body(BodyExtractors.toMono(elementClass));
+		Mono<T> mono = body(BodyExtractors.toMono(elementClass));
+		return mono.mapError(UnsupportedMediaTypeException.class, ERROR_MAPPER);
 	}
 
 	@Override
 	public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
-		return body(BodyExtractors.toFlux(elementClass));
+		Flux<T> flux = body(BodyExtractors.toFlux(elementClass));
+		return flux.mapError(UnsupportedMediaTypeException.class, ERROR_MAPPER);
 	}
 
 	@Override

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/server/DefaultServerRequestTests.java

@@ -32,6 +32,7 @@ import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.buffer.DataBuffer;
@@ -48,6 +49,7 @@ import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.UnsupportedMediaTypeStatusException;
 import org.springframework.web.server.WebSession;
 
 import static org.junit.Assert.assertEquals;
@@ -235,4 +237,25 @@ public class DefaultServerRequestTests {
 		assertEquals(Collections.singletonList("foo"), result.block());
 	}
 
+	@Test
+	public void bodyUnacceptable() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockRequest.getHeaders()).thenReturn(httpHeaders);
+		when(mockRequest.getBody()).thenReturn(body);
+
+		Set<HttpMessageReader<?>> messageReaders = Collections.emptySet();
+		when(mockHandlerStrategies.messageReaders()).thenReturn(messageReaders::stream);
+
+		Flux<String> resultFlux = defaultRequest.bodyToFlux(String.class);
+		StepVerifier.create(resultFlux)
+				.expectError(UnsupportedMediaTypeStatusException.class)
+				.verify();
+	}
+
 }