Add body methods with Object parameter to WebFlux

The commit deprecates syncBody(Object) in favor of body(Object)
which has the same behavior in ServerResponse, WebClient and
WebTestClient. It also adds body(Object, Class) and
body(Object, ParameterizedTypeReference) methods in order to support
any reactive type that can be adapted to a Publisher via
ReactiveAdapterRegistry. Related BodyInserters#fromProducer
methods are provided as well.

Shadowed Kotlin body<T>() extensions are deprecated in favor of
bodyWithType<T>() ones, including dedicated Publisher<T> and
Flow<T> variants. Coroutines extensions are adapted as well, and
body(Object) can now be used with suspending functions.

Closes gh-23212

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

@@ -30,6 +30,7 @@ import java.util.Map;
 import java.util.Optional;
 
 import com.fasterxml.jackson.annotation.JsonView;
+import io.reactivex.Single;
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
@@ -159,6 +160,51 @@ public class BodyInsertersTests {
 				.verify();
 	}
 
+	@Test
+	public void ofProducerWithMono() {
+		Mono<User> body = Mono.just(new User("foo", "bar"));
+		BodyInserter<?, ReactiveHttpOutputMessage> inserter = BodyInserters.fromProducer(body, User.class);
+
+		MockServerHttpResponse response = new MockServerHttpResponse();
+		Mono<Void> result = inserter.insert(response, this.context);
+		StepVerifier.create(result).expectComplete().verify();
+		StepVerifier.create(response.getBodyAsString())
+				.expectNext("{\"username\":\"foo\",\"password\":\"bar\"}")
+				.expectComplete()
+				.verify();
+	}
+
+	@Test
+	public void ofProducerWithFlux() {
+		Flux<String> body = Flux.just("foo");
+		BodyInserter<?, ReactiveHttpOutputMessage> inserter = BodyInserters.fromProducer(body, String.class);
+
+		MockServerHttpResponse response = new MockServerHttpResponse();
+		Mono<Void> result = inserter.insert(response, this.context);
+		StepVerifier.create(result).expectComplete().verify();
+		StepVerifier.create(response.getBody())
+				.consumeNextWith(buf -> {
+					String actual = DataBufferTestUtils.dumpString(buf, UTF_8);
+					assertThat(actual).isEqualTo("foo");
+				})
+				.expectComplete()
+				.verify();
+	}
+
+	@Test
+	public void ofProducerWithSingle() {
+		Single<User> body = Single.just(new User("foo", "bar"));
+		BodyInserter<?, ReactiveHttpOutputMessage> inserter = BodyInserters.fromProducer(body, User.class);
+
+		MockServerHttpResponse response = new MockServerHttpResponse();
+		Mono<Void> result = inserter.insert(response, this.context);
+		StepVerifier.create(result).expectComplete().verify();
+		StepVerifier.create(response.getBodyAsString())
+				.expectNext("{\"username\":\"foo\",\"password\":\"bar\"}")
+				.expectComplete()
+				.verify();
+	}
+
 	@Test
 	public void ofPublisher() {
 		Flux<String> body = Flux.just("foo");

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

@@ -61,7 +61,7 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 		Mono<ClientResponse> result = webClient
 				.post()
 				.uri("http://localhost:" + this.port + "/multipartData")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.exchange();
 
 		StepVerifier
@@ -75,7 +75,7 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 		Mono<ClientResponse> result = webClient
 				.post()
 				.uri("http://localhost:" + this.port + "/parts")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.exchange();
 
 		StepVerifier
@@ -89,7 +89,7 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 		Mono<String> result = webClient
 				.post()
 				.uri("http://localhost:" + this.port + "/transferTo")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToMono(String.class);
 
@@ -169,7 +169,7 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 							Path tempFile = Files.createTempFile("MultipartIntegrationTests", null);
 							return part.transferTo(tempFile)
 									.then(ServerResponse.ok()
-											.syncBody(tempFile.toString()));
+											.body(tempFile.toString()));
 						}
 						catch (Exception e) {
 							return Mono.error(e);

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

@@ -186,7 +186,7 @@ public class DefaultWebClientTests {
 		WebClient client = this.builder.build();
 
 		assertThatIllegalArgumentException().isThrownBy(() ->
-				client.post().uri("https://example.com").syncBody(mono));
+				client.post().uri("https://example.com").body(mono));
 	}
 
 	@Test

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

@@ -354,7 +354,7 @@ public class WebClientIntegrationTests {
 				.uri("/pojo/capitalize")
 				.accept(MediaType.APPLICATION_JSON)
 				.contentType(MediaType.APPLICATION_JSON)
-				.syncBody(new Pojo("foofoo", "barbar"))
+				.body(new Pojo("foofoo", "barbar"))
 				.retrieve()
 				.bodyToMono(Pojo.class);
 

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

@@ -24,6 +24,7 @@ import java.util.EnumSet;
 import java.util.List;
 import java.util.Set;
 
+import io.reactivex.Single;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -76,6 +77,14 @@ public class DefaultEntityResponseBuilderTests {
 		assertThat(response.entity()).isSameAs(body);
 	}
 
+	@Test
+	public void fromProducer() {
+		Single<String> body = Single.just("foo");
+		ParameterizedTypeReference<String> typeReference = new ParameterizedTypeReference<String>() {};
+		EntityResponse<Single<String>> response = EntityResponse.fromProducer(body, typeReference).build().block();
+		assertThat(response.entity()).isSameAs(body);
+	}
+
 	@Test
 	public void status() {
 		String body = "foo";

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

@@ -308,7 +308,7 @@ public class DefaultServerResponseBuilderTests {
 	public void copyCookies() {
 		Mono<ServerResponse> serverResponse = ServerResponse.ok()
 				.cookie(ResponseCookie.from("foo", "bar").build())
-				.syncBody("body");
+				.body("body");
 
 		assertThat(serverResponse.block().cookies().isEmpty()).isFalse();
 
@@ -360,7 +360,7 @@ public class DefaultServerResponseBuilderTests {
 		Mono<Void> mono = Mono.empty();
 
 		assertThatIllegalArgumentException().isThrownBy(() ->
-				ServerResponse.ok().syncBody(mono));
+				ServerResponse.ok().body(mono));
 	}
 
 	@Test
@@ -368,7 +368,7 @@ public class DefaultServerResponseBuilderTests {
 		String etag = "\"foo\"";
 		ServerResponse responseMono = ServerResponse.ok()
 				.eTag(etag)
-				.syncBody("bar")
+				.body("bar")
 				.block();
 
 		MockServerHttpRequest request = MockServerHttpRequest.get("https://example.com")
@@ -392,7 +392,7 @@ public class DefaultServerResponseBuilderTests {
 
 		ServerResponse responseMono = ServerResponse.ok()
 				.lastModified(oneMinuteBeforeNow)
-				.syncBody("bar")
+				.body("bar")
 				.block();
 
 		MockServerHttpRequest request = MockServerHttpRequest.get("https://example.com")

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

@@ -33,8 +33,8 @@ public class InvalidHttpMethodIntegrationTests extends AbstractRouterFunctionInt
 	@Override
 	protected RouterFunction<?> routerFunction() {
 		return RouterFunctions.route(RequestPredicates.GET("/"),
-				request -> ServerResponse.ok().syncBody("FOO"))
-				.andRoute(RequestPredicates.all(), request -> ServerResponse.ok().syncBody("BAR"));
+				request -> ServerResponse.ok().body("FOO"))
+				.andRoute(RequestPredicates.all(), request -> ServerResponse.ok().body("BAR"));
 	}
 
 	@Test

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

@@ -125,7 +125,7 @@ public class NestedRouteIntegrationTests extends AbstractRouterFunctionIntegrati
 
 		public Mono<ServerResponse> pattern(ServerRequest request) {
 			String pattern = matchingPattern(request).getPatternString();
-			return ServerResponse.ok().syncBody(pattern);
+			return ServerResponse.ok().body(pattern);
 		}
 
 		@SuppressWarnings("unchecked")

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

@@ -85,7 +85,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Mono<ClientResponse> result = webClient
 				.post()
 				.uri("/requestPart")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.exchange();
 
 		StepVerifier
@@ -99,7 +99,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Mono<String> result = webClient
 				.post()
 				.uri("/requestBodyMap")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToMono(String.class);
 
@@ -113,7 +113,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Mono<String> result = webClient
 				.post()
 				.uri("/requestBodyFlux")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToMono(String.class);
 
@@ -127,7 +127,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Mono<String> result = webClient
 				.post()
 				.uri("/filePartFlux")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToMono(String.class);
 
@@ -141,7 +141,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Mono<String> result = webClient
 				.post()
 				.uri("/filePartMono")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToMono(String.class);
 
@@ -155,7 +155,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Flux<String> result = webClient
 				.post()
 				.uri("/transferTo")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToFlux(String.class);
 
@@ -183,7 +183,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
 		Mono<String> result = webClient
 				.post()
 				.uri("/modelAttribute")
-				.syncBody(generateBody())
+				.body(generateBody())
 				.retrieve()
 				.bodyToMono(String.class);
 

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt

@@ -27,6 +27,7 @@ import org.junit.Test
 import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
 import reactor.core.publisher.Mono
+import java.util.concurrent.CompletableFuture
 
 /**
  * Mock object based tests for [WebClient] Kotlin extensions
@@ -41,9 +42,9 @@ class WebClientExtensionsTests {
 
 
 	@Test
-	fun `RequestBodySpec#body with Publisher and reified type parameters`() {
+	fun `RequestBodySpec#bodyWithType with Publisher and reified type parameters`() {
 		val body = mockk<Publisher<List<Foo>>>()
-		requestBodySpec.body(body)
+		requestBodySpec.bodyWithType(body)
 		verify { requestBodySpec.body(body, object : ParameterizedTypeReference<List<Foo>>() {}) }
 	}
 
@@ -51,8 +52,16 @@ class WebClientExtensionsTests {
 	@FlowPreview
 	fun `RequestBodySpec#body with Flow and reified type parameters`() {
 		val body = mockk<Flow<List<Foo>>>()
-		requestBodySpec.body(body)
-		verify { requestBodySpec.body(ofType<Publisher<List<Foo>>>(), object : ParameterizedTypeReference<List<Foo>>() {}) }
+		requestBodySpec.bodyWithType(body)
+		verify { requestBodySpec.body(ofType<Any>(), object : ParameterizedTypeReference<List<Foo>>() {}) }
+	}
+
+	@Test
+	@FlowPreview
+	fun `RequestBodySpec#body with CompletableFuture and reified type parameters`() {
+		val body = mockk<CompletableFuture<List<Foo>>>()
+		requestBodySpec.bodyWithType<List<Foo>>(body)
+		verify { requestBodySpec.body(ofType<Any>(), object : ParameterizedTypeReference<List<Foo>>() {}) }
 	}
 
 	@Test
@@ -83,19 +92,6 @@ class WebClientExtensionsTests {
 		}
 	}
 
-	@Test
-	fun body() {
-		val headerSpec = mockk<WebClient.RequestHeadersSpec<*>>()
-		val supplier: suspend () -> String = mockk()
-		every { requestBodySpec.body(ofType<Mono<String>>()) } returns headerSpec
-		runBlocking {
-			requestBodySpec.body(supplier)
-		}
-		verify {
-			requestBodySpec.body(ofType<Mono<String>>())
-		}
-	}
-
 	@Test
 	fun awaitBody() {
 		val spec = mockk<WebClient.ResponseSpec>()

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensionsTests.kt

@@ -19,6 +19,7 @@ package org.springframework.web.reactive.function.server
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import io.reactivex.Flowable
 import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.runBlocking
@@ -28,12 +29,14 @@ import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
 import org.springframework.http.MediaType.*
 import reactor.core.publisher.Mono
+import java.util.concurrent.CompletableFuture
 
 /**
  * Mock object based tests for [ServerResponse] Kotlin extensions
  *
  * @author Sebastien Deleuze
  */
+@Suppress("UnassignedFluxMonoInstance")
 class ServerResponseExtensionsTests {
 
 	private val bodyBuilder = mockk<ServerResponse.BodyBuilder>(relaxed = true)
@@ -42,10 +45,51 @@ class ServerResponseExtensionsTests {
 	@Test
 	fun `BodyBuilder#body with Publisher and reified type parameters`() {
 		val body = mockk<Publisher<List<Foo>>>()
-		bodyBuilder.body(body)
+		bodyBuilder.bodyWithType(body)
 		verify { bodyBuilder.body(body, object : ParameterizedTypeReference<List<Foo>>() {}) }
 	}
 
+	@Test
+	fun `BodyBuilder#body with CompletableFuture and reified type parameters`() {
+		val body = mockk<CompletableFuture<List<Foo>>>()
+		bodyBuilder.bodyWithType<List<Foo>>(body)
+		verify { bodyBuilder.body(body, object : ParameterizedTypeReference<List<Foo>>() {}) }
+	}
+
+	@Test
+	fun `BodyBuilder#body with Flowable and reified type parameters`() {
+		val body = mockk<Flowable<List<Foo>>>()
+		bodyBuilder.bodyWithType(body)
+		verify { bodyBuilder.body(body, object : ParameterizedTypeReference<List<Foo>>() {}) }
+	}
+
+	@Test
+	fun `BodyBuilder#bodyAndAwait with object parameter`() {
+		val response = mockk<ServerResponse>()
+		val body = "foo"
+		every { bodyBuilder.body(ofType<String>()) } returns Mono.just(response)
+		runBlocking {
+			bodyBuilder.bodyAndAwait(body)
+		}
+		verify {
+			bodyBuilder.body(ofType<String>())
+		}
+	}
+
+	@Test
+	@FlowPreview
+	fun `BodyBuilder#bodyAndAwait with flow parameter`() {
+		val response = mockk<ServerResponse>()
+		val body = mockk<Flow<List<Foo>>>()
+		every { bodyBuilder.body(ofType<Flow<List<Foo>>>(), object : ParameterizedTypeReference<List<Foo>>() {}) } returns Mono.just(response)
+		runBlocking {
+			bodyBuilder.bodyAndAwait(body)
+		}
+		verify {
+			bodyBuilder.body(ofType<Flow<List<Foo>>>(), object : ParameterizedTypeReference<List<Foo>>() {})
+		}
+	}
+
 	@Test
 	fun `BodyBuilder#json`() {
 		bodyBuilder.json()
@@ -71,42 +115,7 @@ class ServerResponseExtensionsTests {
 	}
 
 	@Test
-	fun await() {
-		val response = mockk<ServerResponse>()
-		val builder = mockk<ServerResponse.HeadersBuilder<*>>()
-		every { builder.build() } returns Mono.just(response)
-		runBlocking {
-			assertEquals(response, builder.buildAndAwait())
-		}
-	}
-
-	@Test
-	fun `bodyAndAwait with object parameter`() {
-		val response = mockk<ServerResponse>()
-		val body = "foo"
-		every { bodyBuilder.syncBody(ofType<String>()) } returns Mono.just(response)
-		runBlocking {
-			bodyBuilder.bodyAndAwait(body)
-		}
-		verify {
-			bodyBuilder.syncBody(ofType<String>())
-		}
-	}
-
-	@Test
-	@FlowPreview
-	fun bodyFlowAndAwait() {
-		val response = mockk<ServerResponse>()
-		val body = mockk<Flow<List<Foo>>>()
-		every { bodyBuilder.body(ofType<Publisher<List<Foo>>>()) } returns Mono.just(response)
-		runBlocking {
-			bodyBuilder.bodyFlowAndAwait(body)
-		}
-		verify { bodyBuilder.body(ofType<Publisher<List<Foo>>>(), object : ParameterizedTypeReference<List<Foo>>() {}) }
-	}
-
-	@Test
-	fun `renderAndAwait with a vararg parameter`() {
+	fun `BodyBuilder#renderAndAwait with a vararg parameter`() {
 		val response = mockk<ServerResponse>()
 		every { bodyBuilder.render("foo", any(), any()) } returns Mono.just(response)
 		runBlocking {
@@ -118,7 +127,7 @@ class ServerResponseExtensionsTests {
 	}
 
 	@Test
-	fun `renderAndAwait with a Map parameter`() {
+	fun `BodyBuilder#renderAndAwait with a Map parameter`() {
 		val response = mockk<ServerResponse>()
 		val map = mockk<Map<String, *>>()
 		every { bodyBuilder.render("foo", map) } returns Mono.just(response)
@@ -130,5 +139,15 @@ class ServerResponseExtensionsTests {
 		}
 	}
 
+	@Test
+	fun `HeadersBuilder#buildAndAwait`() {
+		val response = mockk<ServerResponse>()
+		val builder = mockk<ServerResponse.HeadersBuilder<*>>()
+		every { builder.build() } returns Mono.just(response)
+		runBlocking {
+			assertEquals(response, builder.buildAndAwait())
+		}
+	}
+
 	class Foo
 }