Allow empty body with no content type in BodyExtractors

Issue: SPR-15758

Author: rstoyanchev

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

@@ -103,7 +103,8 @@ static <T> BodyExtractor<Mono<T>, ReactiveHttpInputMessage> toMono(ResolvableTyp
 						return reader.readMono(elementType, inputMessage, context.hints());
 					}
 				},
-				Mono::error,
+				ex -> (inputMessage.getHeaders().getContentType() == null) ?
+						Mono.from(permitEmptyOrFail(inputMessage, ex)) : Mono.error(ex),
 				Mono::empty);
 	}
 
@@ -138,6 +139,7 @@ public static <T> BodyExtractor<Flux<T>, ReactiveHttpInputMessage> toFlux(Parame
 		return toFlux(ResolvableType.forType(typeReference.getType()));
 	}
 
+	@SuppressWarnings("unchecked")
 	static <T> BodyExtractor<Flux<T>, ReactiveHttpInputMessage> toFlux(ResolvableType elementType) {
 		Assert.notNull(elementType, "'elementType' must not be null");
 		return (inputMessage, context) -> readWithMessageReaders(inputMessage, context,
@@ -152,10 +154,18 @@ static <T> BodyExtractor<Flux<T>, ReactiveHttpInputMessage> toFlux(ResolvableTyp
 						return reader.read(elementType, inputMessage, context.hints());
 					}
 				},
-				Flux::error,
+				ex -> (inputMessage.getHeaders().getContentType() == null) ?
+						permitEmptyOrFail(inputMessage, ex) : Flux.error(ex),
 				Flux::empty);
 	}
 
+	@SuppressWarnings("unchecked")
+	private static <T> Flux<T> permitEmptyOrFail(ReactiveHttpInputMessage message, UnsupportedMediaTypeException ex) {
+		return message.getBody().doOnNext(buffer -> {
+			throw ex;
+		}).map(o -> (T) o);
+	}
+
 	/**
 	 * Return a {@code BodyExtractor} that reads form data into a {@link MultiValueMap}.
 	 * @return a {@code BodyExtractor} that reads form data
@@ -225,7 +235,8 @@ public static BodyExtractor<Flux<DataBuffer>, ReactiveHttpInputMessage> toDataBu
 
 	private static <T, S extends Publisher<T>> S readWithMessageReaders(
 			ReactiveHttpInputMessage inputMessage, BodyExtractor.Context context, ResolvableType elementType,
-			Function<HttpMessageReader<T>, S> readerFunction, Function<Throwable, S> unsupportedError,
+			Function<HttpMessageReader<T>, S> readerFunction,
+			Function<UnsupportedMediaTypeException, S> unsupportedError,
 			Supplier<S> empty) {
 
 		if (VOID_TYPE.equals(elementType)) {

spring-webflux/src/test/java/org/springframework/web/reactive/function/BodyExtractorsTests.java

@@ -56,7 +56,9 @@
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.util.MultiValueMap;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 import static org.springframework.http.codec.json.Jackson2CodecSupport.JSON_VIEW_HINT;
 
 /**
@@ -169,6 +171,17 @@ public void toMonoWithHints() throws Exception {
 				.verify();
 	}
 
+	@Test // SPR-15758
+	public void toMonoWithEmptyBodyAndNoContentType() throws Exception {
+		BodyExtractor<Mono<Map<String, String>>, ReactiveHttpInputMessage> extractor =
+				BodyExtractors.toMono(new ParameterizedTypeReference<Map<String, String>>() {});
+
+		MockServerHttpRequest request = MockServerHttpRequest.post("/").body(Flux.empty());
+		Mono<Map<String, String>> result = extractor.extract(request, this.context);
+
+		StepVerifier.create(result).expectComplete().verify();
+	}
+
 	@Test
 	public void toFlux() throws Exception {
 		BodyExtractor<Flux<String>, ReactiveHttpInputMessage> extractor = BodyExtractors.toFlux(String.class);