Comprensive support for empty request body

This commit adds support for handling an empty request body with both
HttpEntity where the body is not required and with @RequestBody where
the body is required depending on the annotation's required flag.

If the body is an explicit type (e.g. String, HttpEntity<String>) and
the body is required an exception is raised before the method is even
invoked or otherwise the body is passed in as null.

If the body is declared as an async type (e.g. Mono<String>,
HttpEntity<Mono<String>>) and is required, the error will flow through
the async type. If not required, the async type will be passed with no
values (i.e. empty).

A notable exception is rx.Single which can only have one value or one
error and cannot be empty. As a result currently the use of rx.Single
to represent the request body in any form effectively implies the body
is required.

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/AbstractMessageConverterArgumentResolver.java

@@ -105,7 +105,8 @@ public abstract class AbstractMessageConverterArgumentResolver {
 	}
 
 
-	protected Mono<Object> readBody(MethodParameter bodyParameter, ServerWebExchange exchange) {
+	protected Mono<Object> readBody(MethodParameter bodyParameter, boolean isBodyRequired,
+			ServerWebExchange exchange) {
 
 		TypeDescriptor typeDescriptor = new TypeDescriptor(bodyParameter);
 		boolean convertFromMono = getConversionService().canConvert(MONO_TYPE, typeDescriptor);
@@ -125,14 +126,22 @@ public abstract class AbstractMessageConverterArgumentResolver {
 		for (HttpMessageConverter<?> converter : getMessageConverters()) {
 			if (converter.canRead(elementType, mediaType)) {
 				if (convertFromFlux) {
-					Flux<?> flux = converter.read(elementType, request);
+					Flux<?> flux = converter.read(elementType, request)
+							.onErrorResumeWith(ex -> Flux.error(getReadError(ex, bodyParameter)));
+					if (checkRequired(bodyParameter, isBodyRequired)) {
+						flux = flux.switchIfEmpty(Flux.error(getRequiredBodyError(bodyParameter)));
+					}
 					if (this.validator != null) {
 						flux = flux.map(applyValidationIfApplicable(bodyParameter));
 					}
 					return Mono.just(getConversionService().convert(flux, FLUX_TYPE, typeDescriptor));
 				}
 				else {
-					Mono<?> mono = converter.readMono(elementType, request);
+					Mono<?> mono = converter.readMono(elementType, request)
+							.otherwise(ex -> Mono.error(getReadError(ex, bodyParameter)));
+					if (checkRequired(bodyParameter, isBodyRequired)) {
+						mono = mono.otherwiseIfEmpty(Mono.error(getRequiredBodyError(bodyParameter)));
+					}
 					if (this.validator != null) {
 						mono = mono.map(applyValidationIfApplicable(bodyParameter));
 					}
@@ -149,6 +158,22 @@ public abstract class AbstractMessageConverterArgumentResolver {
 		return Mono.error(new UnsupportedMediaTypeStatusException(mediaType, this.supportedMediaTypes));
 	}
 
+	protected boolean checkRequired(MethodParameter bodyParameter, boolean isBodyRequired) {
+		if ("rx.Single".equals(bodyParameter.getNestedParameterType().getName())) {
+			return true;
+		}
+		return isBodyRequired;
+	}
+
+	protected ServerWebInputException getReadError(Throwable ex, MethodParameter parameter) {
+		return new ServerWebInputException("Failed to read HTTP message", parameter, ex);
+	}
+
+	protected ServerWebInputException getRequiredBodyError(MethodParameter parameter) {
+		return new ServerWebInputException("Required request body is missing: " +
+				parameter.getMethod().toGenericString());
+	}
+
 	protected <T> Function<T, T> applyValidationIfApplicable(MethodParameter methodParam) {
 		Annotation[] annotations = methodParam.getParameterAnnotations();
 		for (Annotation ann : annotations) {

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/HttpEntityArgumentResolver.java

@@ -91,17 +91,22 @@ public class HttpEntityArgumentResolver extends AbstractMessageConverterArgument
 			bodyParameter.increaseNestingLevel();
 		}
 
-		return readBody(bodyParameter, exchange)
-				.map(body -> {
-					ServerHttpRequest request = exchange.getRequest();
-					HttpHeaders headers = request.getHeaders();
-					if (RequestEntity.class == entityType.getRawClass()) {
-						return new RequestEntity<>(body, headers, request.getMethod(), request.getURI());
-					}
-					else {
-						return new HttpEntity<>(body, headers);
-					}
-				});
+		return readBody(bodyParameter, false, exchange)
+				.map(body -> createHttpEntity(body, entityType, exchange))
+				.defaultIfEmpty(createHttpEntity(null, entityType, exchange));
+	}
+
+	private Object createHttpEntity(Object body, ResolvableType entityType,
+			ServerWebExchange exchange) {
+
+		ServerHttpRequest request = exchange.getRequest();
+		HttpHeaders headers = request.getHeaders();
+		if (RequestEntity.class == entityType.getRawClass()) {
+			return new RequestEntity<>(body, headers, request.getMethod(), request.getURI());
+		}
+		else {
+			return new HttpEntity<>(body, headers);
+		}
 	}
 
 }

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolver.java

@@ -21,7 +21,6 @@ import java.util.List;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.MethodParameter;
-import org.springframework.core.ResolvableType;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.http.converter.reactive.HttpMessageConverter;
 import org.springframework.ui.ModelMap;
@@ -79,7 +78,8 @@ public class RequestBodyArgumentResolver extends AbstractMessageConverterArgumen
 
 	@Override
 	public Mono<Object> resolveArgument(MethodParameter param, ModelMap model, ServerWebExchange exchange) {
-		return readBody(param, exchange);
+		boolean isRequired = param.getParameterAnnotation(RequestBody.class).required();
+		return readBody(param, isRequired, exchange);
 	}
 
 }

spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java

@@ -49,6 +49,7 @@ import org.springframework.web.reactive.result.method.annotation.ResponseBodyRes
 import org.springframework.web.server.NotAcceptableStatusException;
 import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.ServerWebInputException;
 import org.springframework.web.server.WebExceptionHandler;
 import org.springframework.web.server.WebHandler;
 import org.springframework.web.server.adapter.DefaultServerWebExchange;
@@ -163,7 +164,8 @@ public class DispatcherHandlerErrorTests {
 		Mono<Void> publisher = this.dispatcherHandler.handle(this.exchange);
 
 		TestSubscriber.subscribe(publisher)
-				.assertErrorWith(ex -> assertSame(EXCEPTION, ex));
+				.assertError(ServerWebInputException.class)
+				.assertErrorWith(ex -> assertSame(EXCEPTION, ex.getCause()));
 	}
 
 	@Test

spring-web-reactive/src/test/java/org/springframework/web/reactive/result/ResolvableMethod.java

@@ -171,8 +171,8 @@ public class ResolvableMethod {
 			matches.add(param);
 		}
 
-		Assert.isTrue(!matches.isEmpty(), "No matching method argument: " + this);
-		Assert.isTrue(matches.size() == 1, "Multiple matching method arguments: " + matches);
+		Assert.isTrue(!matches.isEmpty(), "No matching arg on " + method.toString());
+		Assert.isTrue(matches.size() == 1, "Multiple matching args: " + matches + " on " + method.toString());
 
 		return matches.get(0);
 	}

spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/HttpEntityArgumentResolverTests.java

@@ -25,6 +25,8 @@ import java.util.concurrent.CompletableFuture;
 
 import org.junit.Before;
 import org.junit.Test;
+import reactor.core.converter.RxJava1ObservableConverter;
+import reactor.core.converter.RxJava1SingleConverter;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.test.TestSubscriber;
@@ -51,12 +53,14 @@ import org.springframework.http.server.reactive.MockServerHttpResponse;
 import org.springframework.ui.ExtendedModelMap;
 import org.springframework.web.reactive.result.ResolvableMethod;
 import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.ServerWebInputException;
 import org.springframework.web.server.adapter.DefaultServerWebExchange;
 import org.springframework.web.server.session.MockWebSessionManager;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.springframework.core.ResolvableType.forClassWithGenerics;
 
@@ -106,6 +110,68 @@ public class HttpEntityArgumentResolverTests {
 		assertFalse(this.resolver.supportsParameter(this.testMethod.resolveParam(type)));
 	}
 
+	@Test
+	public void emptyBodyWithString() throws Exception {
+		ResolvableType type = httpEntity(String.class);
+		HttpEntity<Object> entity = resolveValueWithEmptyBody(type);
+
+		assertNull(entity.getBody());
+	}
+
+	@Test
+	public void emptyBodyWithMono() throws Exception {
+		ResolvableType type = httpEntity(forClassWithGenerics(Mono.class, String.class));
+		HttpEntity<Mono<String>> entity = resolveValueWithEmptyBody(type);
+
+		TestSubscriber.subscribe(entity.getBody())
+				.assertNoError()
+				.assertComplete()
+				.assertNoValues();
+	}
+
+	@Test
+	public void emptyBodyWithFlux() throws Exception {
+		ResolvableType type = httpEntity(forClassWithGenerics(Flux.class, String.class));
+		HttpEntity<Flux<String>> entity = resolveValueWithEmptyBody(type);
+
+		TestSubscriber.subscribe(entity.getBody())
+				.assertNoError()
+				.assertComplete()
+				.assertNoValues();
+	}
+
+	@Test
+	public void emptyBodyWithSingle() throws Exception {
+		ResolvableType type = httpEntity(forClassWithGenerics(Single.class, String.class));
+		HttpEntity<Single<String>> entity = resolveValueWithEmptyBody(type);
+
+		TestSubscriber.subscribe(RxJava1SingleConverter.from(entity.getBody()))
+				.assertNoValues()
+				.assertError(ServerWebInputException.class);
+	}
+
+	@Test
+	public void emptyBodyWithObservable() throws Exception {
+		ResolvableType type = httpEntity(forClassWithGenerics(Observable.class, String.class));
+		HttpEntity<Observable<String>> entity = resolveValueWithEmptyBody(type);
+
+		TestSubscriber.subscribe(RxJava1ObservableConverter.from(entity.getBody()))
+				.assertNoError()
+				.assertComplete()
+				.assertNoValues();
+	}
+
+	@Test
+	public void emptyBodyWithCompletableFuture() throws Exception {
+		ResolvableType type = httpEntity(forClassWithGenerics(CompletableFuture.class, String.class));
+		HttpEntity<CompletableFuture<String>> entity = resolveValueWithEmptyBody(type);
+
+		entity.getBody().whenComplete((body, ex) -> {
+			assertNull(body);
+			assertNull(ex);
+		});
+	}
+
 	@Test
 	public void httpEntityWithStringBody() throws Exception {
 		String body = "line1";
@@ -211,6 +277,17 @@ public class HttpEntityArgumentResolverTests {
 		return (T) value;
 	}
 
+	@SuppressWarnings("unchecked")
+	private <T> HttpEntity<T> resolveValueWithEmptyBody(ResolvableType type) {
+		this.request.writeWith(Flux.empty());
+		MethodParameter param = this.testMethod.resolveParam(type);
+		Mono<Object> result = this.resolver.resolveArgument(param, new ExtendedModelMap(), this.exchange);
+		HttpEntity<String> httpEntity = (HttpEntity<String>) result.block(Duration.ofSeconds(5));
+
+		assertEquals(this.request.getHeaders(), httpEntity.getHeaders());
+		return (HttpEntity<T>) httpEntity;
+	}
+
 	private DataBuffer dataBuffer(String body) {
 		byte[] bytes = body.getBytes(Charset.forName("UTF-8"));
 		ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
@@ -224,10 +301,10 @@ public class HttpEntityArgumentResolverTests {
 			Mono<String> monoString,
 			HttpEntity<String> httpEntity,
 			HttpEntity<Mono<String>> monoBody,
-			HttpEntity<Single<String>> singleBody,
-			HttpEntity<CompletableFuture<String>> completableFutureBody,
 			HttpEntity<Flux<String>> fluxBody,
+			HttpEntity<Single<String>> singleBody,
 			HttpEntity<Observable<String>> observableBody,
+			HttpEntity<CompletableFuture<String>> completableFutureBody,
 			RequestEntity<String> requestEntity) {}
 
 }

spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageConverterArgumentResolverTests.java

@@ -88,7 +88,7 @@ public class MessageConverterArgumentResolverTests {
 
 	@Before
 	public void setUp() throws Exception {
-		this.request = new MockServerHttpRequest(HttpMethod.GET, new URI("/path"));
+		this.request = new MockServerHttpRequest(HttpMethod.POST, new URI("/path"));
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		this.exchange = new DefaultServerWebExchange(this.request, response, new MockWebSessionManager());
 	}
@@ -100,20 +100,23 @@ public class MessageConverterArgumentResolverTests {
 		this.request.writeWith(Flux.just(dataBuffer(body)));
 		ResolvableType type = forClassWithGenerics(Mono.class, TestBean.class);
 		MethodParameter param = this.testMethod.resolveParam(type);
-		Mono<Object> result = this.resolver.readBody(param, this.exchange);
+		Mono<Object> result = this.resolver.readBody(param, true, this.exchange);
 
 		TestSubscriber.subscribe(result)
 				.assertError(UnsupportedMediaTypeStatusException.class);
 	}
 
-	@Test // SPR-9942
-	public void noContent() throws Exception {
+	// More extensive "empty body" tests in RequestBody- and HttpEntityArgumentResolverTests
+
+	@Test @SuppressWarnings("unchecked") // SPR-9942
+	public void emptyBody() throws Exception {
 		this.request.writeWith(Flux.empty());
+		this.request.getHeaders().setContentType(MediaType.APPLICATION_JSON);
 		ResolvableType type = forClassWithGenerics(Mono.class, TestBean.class);
 		MethodParameter param = this.testMethod.resolveParam(type);
-		Mono<Object> result = this.resolver.readBody(param, this.exchange);
+		Mono<TestBean> result = (Mono<TestBean>) this.resolver.readBody(param, true, this.exchange).block();
 
-		TestSubscriber.subscribe(result).assertError(UnsupportedMediaTypeStatusException.class);
+		TestSubscriber.subscribe(result).assertError(ServerWebInputException.class);
 	}
 
 	@Test
@@ -262,7 +265,7 @@ public class MessageConverterArgumentResolverTests {
 		this.request.getHeaders().setContentType(MediaType.APPLICATION_JSON);
 		this.request.writeWith(Flux.just(dataBuffer(body)));
 
-		Mono<Object> result = this.resolver.readBody(param, this.exchange);
+		Mono<Object> result = this.resolver.readBody(param, true, this.exchange);
 		Object value = result.block(Duration.ofSeconds(5));
 
 		assertNotNull(value);

spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolverTests.java

@@ -15,26 +15,53 @@
  */
 package org.springframework.web.reactive.result.method.annotation;
 
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
 
+import org.junit.Before;
 import org.junit.Test;
+import reactor.core.converter.RxJava1ObservableConverter;
+import reactor.core.converter.RxJava1SingleConverter;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.test.TestSubscriber;
+import rx.Observable;
+import rx.Single;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.convert.support.MonoToCompletableFutureConverter;
 import org.springframework.core.convert.support.ReactorToRxJava1Converter;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.format.support.DefaultFormattingConversionService;
 import org.springframework.format.support.FormattingConversionService;
+import org.springframework.http.HttpMethod;
 import org.springframework.http.converter.reactive.CodecHttpMessageConverter;
 import org.springframework.http.converter.reactive.HttpMessageConverter;
+import org.springframework.http.server.reactive.MockServerHttpRequest;
+import org.springframework.http.server.reactive.MockServerHttpResponse;
+import org.springframework.ui.ExtendedModelMap;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.reactive.result.ResolvableMethod;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.ServerWebInputException;
+import org.springframework.web.server.adapter.DefaultServerWebExchange;
+import org.springframework.web.server.session.MockWebSessionManager;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.springframework.core.ResolvableType.forClass;
 import static org.springframework.core.ResolvableType.forClassWithGenerics;
 
 /**
@@ -46,21 +73,130 @@ import static org.springframework.core.ResolvableType.forClassWithGenerics;
  */
 public class RequestBodyArgumentResolverTests {
 
+	private RequestBodyArgumentResolver resolver = resolver();
+
+	private ServerWebExchange exchange;
+
+	private MockServerHttpRequest request;
+
+	private ResolvableMethod testMethod = ResolvableMethod.on(this.getClass()).name("handle");
+
+
+	@Before
+	public void setUp() throws Exception {
+		this.request = new MockServerHttpRequest(HttpMethod.POST, new URI("/path"));
+		MockServerHttpResponse response = new MockServerHttpResponse();
+		this.exchange = new DefaultServerWebExchange(this.request, response, new MockWebSessionManager());
+	}
+
 
 	@Test
 	public void supports() throws Exception {
+		ResolvableType type = forClassWithGenerics(Mono.class, String.class);
+		MethodParameter param = this.testMethod.resolveParam(type, requestBody(true));
+		assertTrue(this.resolver.supportsParameter(param));
+
+		MethodParameter parameter = this.testMethod.resolveParam(p -> !p.hasParameterAnnotations());
+		assertFalse(this.resolver.supportsParameter(parameter));
+	}
 
-		ResolvableMethod testMethod = ResolvableMethod.on(getClass()).name("handle");
-		RequestBodyArgumentResolver resolver = resolver();
+	@Test
+	public void stringBody() throws Exception {
+		String body = "line1";
+		ResolvableType type = forClass(String.class);
+		MethodParameter param = this.testMethod.resolveParam(type, requestBody(true));
+		String value = resolveValue(param, body);
 
+		assertEquals(body, value);
+	}
+
+	@Test(expected = ServerWebInputException.class)
+	public void emptyBodyWithString() throws Exception {
+		resolveValueWithEmptyBody(forClass(String.class), true);
+	}
+
+	@Test
+	public void emptyBodyWithStringNotRequired() throws Exception {
+		ResolvableType type = forClass(String.class);
+		String body = resolveValueWithEmptyBody(type, false);
+
+		assertNull(body);
+	}
+
+	@Test
+	public void emptyBodyWithMono() throws Exception {
 		ResolvableType type = forClassWithGenerics(Mono.class, String.class);
-		MethodParameter param = testMethod.resolveParam(type);
-		assertTrue(resolver.supportsParameter(param));
 
-		MethodParameter parameter = testMethod.resolveParam(p -> !p.hasParameterAnnotations());
-		assertFalse(resolver.supportsParameter(parameter));
+		TestSubscriber.subscribe(resolveValueWithEmptyBody(type, true))
+				.assertNoValues()
+				.assertError(ServerWebInputException.class);
+
+		TestSubscriber.subscribe(resolveValueWithEmptyBody(type, false))
+				.assertNoValues()
+				.assertComplete();
+	}
+
+	@Test
+	public void emptyBodyWithFlux() throws Exception {
+		ResolvableType type = forClassWithGenerics(Flux.class, String.class);
+
+		TestSubscriber.subscribe(resolveValueWithEmptyBody(type, true))
+				.assertNoValues()
+				.assertError(ServerWebInputException.class);
+
+		TestSubscriber.subscribe(resolveValueWithEmptyBody(type, false))
+				.assertNoValues()
+				.assertComplete();
+	}
+
+	@Test
+	public void emptyBodyWithSingle() throws Exception {
+		ResolvableType type = forClassWithGenerics(Single.class, String.class);
+
+		Single<String> single = resolveValueWithEmptyBody(type, true);
+		TestSubscriber.subscribe(RxJava1SingleConverter.from(single))
+				.assertNoValues()
+				.assertError(ServerWebInputException.class);
+
+		single = resolveValueWithEmptyBody(type, false);
+		TestSubscriber.subscribe(RxJava1SingleConverter.from(single))
+				.assertNoValues()
+				.assertError(ServerWebInputException.class);
+	}
+
+	@Test
+	public void emptyBodyWithObservable() throws Exception {
+		ResolvableType type = forClassWithGenerics(Observable.class, String.class);
+
+		Observable<String> observable = resolveValueWithEmptyBody(type, true);
+		TestSubscriber.subscribe(RxJava1ObservableConverter.from(observable))
+				.assertNoValues()
+				.assertError(ServerWebInputException.class);
+
+		observable = resolveValueWithEmptyBody(type, false);
+		TestSubscriber.subscribe(RxJava1ObservableConverter.from(observable))
+				.assertNoValues()
+				.assertComplete();
+	}
+
+	@Test
+	public void emptyBodyWithCompletableFuture() throws Exception {
+		ResolvableType type = forClassWithGenerics(CompletableFuture.class, String.class);
+
+		CompletableFuture<String> future = resolveValueWithEmptyBody(type, true);
+		future.whenComplete((text, ex) -> {
+			assertNull(text);
+			assertNotNull(ex);
+		});
+
+		future = resolveValueWithEmptyBody(type, false);
+		future.whenComplete((text, ex) -> {
+			assertNotNull(text);
+			assertNull(ex);
+		});
 	}
 
+
 	private RequestBodyArgumentResolver resolver() {
 		List<HttpMessageConverter<?>> converters = new ArrayList<>();
 		converters.add(new CodecHttpMessageConverter<>(new StringDecoder()));
@@ -72,8 +208,62 @@ public class RequestBodyArgumentResolverTests {
 		return new RequestBodyArgumentResolver(converters, service);
 	}
 
+	private <T> T resolveValue(MethodParameter param, String body) {
+		this.request.writeWith(Flux.just(dataBuffer(body)));
+		Mono<Object> result = this.resolver.readBody(param, true, this.exchange);
+		Object value = result.block(Duration.ofSeconds(5));
+
+		assertNotNull(value);
+		assertTrue("Unexpected return value type: " + value,
+				param.getParameterType().isAssignableFrom(value.getClass()));
+
+		//noinspection unchecked
+		return (T) value;
+	}
+
+	private <T> T resolveValueWithEmptyBody(ResolvableType type, boolean required) {
+		this.request.writeWith(Flux.empty());
+		MethodParameter param = this.testMethod.resolveParam(type, requestBody(required));
+		Mono<Object> result = this.resolver.resolveArgument(param, new ExtendedModelMap(), this.exchange);
+		Object value = result.block(Duration.ofSeconds(5));
+
+		if (value != null) {
+			assertTrue("Unexpected return value type: " + value,
+					param.getParameterType().isAssignableFrom(value.getClass()));
+		}
+
+		//noinspection unchecked
+		return (T) value;
+	}
+
+	private Predicate<MethodParameter> requestBody(boolean required) {
+		return p -> {
+			RequestBody annotation = p.getParameterAnnotation(RequestBody.class);
+			return annotation != null && annotation.required() == required;
+		};
+	}
+
+	private DataBuffer dataBuffer(String body) {
+		byte[] bytes = body.getBytes(Charset.forName("UTF-8"));
+		ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+		return new DefaultDataBufferFactory().wrap(byteBuffer);
+	}
+
 
 	@SuppressWarnings("unused")
-	void handle(@RequestBody Mono<String> monoString, String paramWithoutAnnotation) {}
+	void handle(
+			@RequestBody String string,
+			@RequestBody Mono<String> mono,
+			@RequestBody Flux<String> flux,
+			@RequestBody Single<String> single,
+			@RequestBody Observable<String> obs,
+			@RequestBody CompletableFuture<String> future,
+			@RequestBody(required = false) String stringNotRequired,
+			@RequestBody(required = false) Mono<String> monoNotRequired,
+			@RequestBody(required = false) Flux<String> fluxNotRequired,
+			@RequestBody(required = false) Single<String> singleNotRequired,
+			@RequestBody(required = false) Observable<String> obsNotRequired,
+			@RequestBody(required = false) CompletableFuture<String> futureNotRequired,
+			String notAnnotated) {}
 
 }