Add Coroutines support for WebClient and WebFlux.fn

This commit is the first part of a more complete Coroutines
support coming in Spring Framework 5.2. It introduces suspendable
Kotlin extensions for Mono based methods in WebFlux classes like
WebClient, ServerRequest, ServerResponse as well as a Coroutines
router usable via `coRouter { }`.

Coroutines extensions use `await` prefix or `AndAwait` suffix,
and most are using names close to their Reactive counterparts,
except `exchange` in `WebClient.RequestHeadersSpec`
which translates to `awaitResponse`.

Upcoming expected changes are:
 - Leverage `Dispatchers.Unconfined` (Kotlin/kotlinx.coroutines#972)
 - Expose extensions for `Flux` based API (Kotlin/kotlinx.coroutines#254)
 - Introduce interop with `CoroutineContext` (Kotlin/kotlinx.coroutines#284)
 - Support Coroutines in `ReactiveAdapterRegistry`
 - Support Coroutines for WebFlux annotated controllers
 - Fix return type of Kotlin suspending functions (gh-21058)

See gh-19975

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

@@ -16,10 +16,16 @@
 
 package org.springframework.web.reactive.function.client
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertEquals
 import org.junit.Test
 import org.springframework.core.ParameterizedTypeReference
+import org.springframework.http.HttpStatus
+import org.springframework.http.ResponseEntity
+import reactor.core.publisher.Mono
 
 /**
  * Mock object based tests for [ClientResponse] Kotlin extensions.
@@ -28,7 +34,7 @@ import org.springframework.core.ParameterizedTypeReference
  */
 class ClientResponseExtensionsTests {
 
-	val response = mockk<ClientResponse>(relaxed = true)
+	private val response = mockk<ClientResponse>(relaxed = true)
 
 	@Test
 	fun `bodyToMono with reified type parameters`() {
@@ -54,5 +60,34 @@ class ClientResponseExtensionsTests {
 		verify { response.toEntityList(object : ParameterizedTypeReference<List<Foo>>() {}) }
 	}
 
+	@Test
+	fun awaitBody() {
+		val response = mockk<ClientResponse>()
+		every { response.bodyToMono<String>() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", response.awaitBody<String>())
+		}
+	}
+
+	@Test
+	fun awaitEntity() {
+		val response = mockk<ClientResponse>()
+		val entity = ResponseEntity("foo", HttpStatus.OK)
+		every { response.toEntity<String>() } returns Mono.just(entity)
+		runBlocking {
+			assertEquals(entity, response.awaitEntity<String>())
+		}
+	}
+
+	@Test
+	fun awaitEntityList() {
+		val response = mockk<ClientResponse>()
+		val entity = ResponseEntity(listOf("foo"), HttpStatus.OK)
+		every { response.toEntityList<String>() } returns Mono.just(entity)
+		runBlocking {
+			assertEquals(entity, response.awaitEntityList<String>())
+		}
+	}
+
 	class Foo
 }

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

@@ -16,11 +16,15 @@
 
 package org.springframework.web.reactive.function.client
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertEquals
 import org.junit.Test
 import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
+import reactor.core.publisher.Mono
 
 /**
  * Mock object based tests for [WebClient] Kotlin extensions
@@ -29,9 +33,9 @@ import org.springframework.core.ParameterizedTypeReference
  */
 class WebClientExtensionsTests {
 
-	val requestBodySpec = mockk<WebClient.RequestBodySpec>(relaxed = true)
+	private val requestBodySpec = mockk<WebClient.RequestBodySpec>(relaxed = true)
 
-	val responseSpec = mockk<WebClient.ResponseSpec>(relaxed = true)
+	private val responseSpec = mockk<WebClient.ResponseSpec>(relaxed = true)
 
 
 	@Test
@@ -53,5 +57,36 @@ class WebClientExtensionsTests {
 		verify { responseSpec.bodyToFlux(object : ParameterizedTypeReference<List<Foo>>() {}) }
 	}
 
+	@Test
+	fun awaitResponse() {
+		val response = mockk<ClientResponse>()
+		every { requestBodySpec.exchange() } returns Mono.just(response)
+		runBlocking {
+			assertEquals(response, requestBodySpec.awaitResponse())
+		}
+	}
+
+	@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>()
+		every { spec.bodyToMono<String>() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", spec.awaitBody<String>())
+		}
+	}
+
 	class Foo
 }

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

@@ -0,0 +1,166 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.reactive.function.server
+
+import org.junit.Test
+import org.springframework.core.io.ClassPathResource
+import org.springframework.http.HttpHeaders.*
+import org.springframework.http.HttpMethod.*
+import org.springframework.http.MediaType.*
+import org.springframework.web.reactive.function.server.MockServerRequest.builder
+import reactor.test.StepVerifier
+import java.net.URI
+
+/**
+ * Tests for [CoRouterFunctionDsl].
+ *
+ * @author Sebastien Deleuze
+ */
+class CoRouterFunctionDslTests {
+
+	@Test
+	fun header() {
+		val request = builder().header("bar", "bar").build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun accept() {
+		val request = builder().uri(URI("/content")).header(ACCEPT, APPLICATION_ATOM_XML_VALUE).build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun acceptAndPOST() {
+		val request = builder()
+				.method(POST)
+				.uri(URI("/api/foo/"))
+				.header(ACCEPT, APPLICATION_JSON_VALUE)
+				.build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun contentType() {
+		val request = builder().uri(URI("/content")).header(CONTENT_TYPE, APPLICATION_OCTET_STREAM_VALUE).build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun resourceByPath() {
+		val request = builder().uri(URI("/org/springframework/web/reactive/function/response.txt")).build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun method() {
+		val request = builder().method(PATCH).build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun path() {
+		val request = builder().uri(URI("/baz")).build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun resource() {
+		val request = builder().uri(URI("/response.txt")).build()
+		StepVerifier.create(sampleRouter().route(request))
+				.expectNextCount(1)
+				.verifyComplete()
+	}
+
+	@Test
+	fun noRoute() {
+		val request = builder()
+				.uri(URI("/bar"))
+				.header(ACCEPT, APPLICATION_PDF_VALUE)
+				.header(CONTENT_TYPE, APPLICATION_PDF_VALUE)
+				.build()
+		StepVerifier.create(sampleRouter().route(request))
+				.verifyComplete()
+	}
+
+	@Test
+	fun rendering() {
+		val request = builder().uri(URI("/rendering")).build()
+		StepVerifier.create(sampleRouter().route(request).flatMap { it.handle(request) })
+				.expectNextMatches { it is RenderingResponse}
+				.verifyComplete()
+	}
+
+	@Test(expected = IllegalStateException::class)
+	fun emptyRouter() {
+		router { }
+	}
+
+
+	private fun sampleRouter() = coRouter {
+		(GET("/foo/") or GET("/foos/")) { req -> handle(req) }
+		"/api".nest {
+			POST("/foo/", ::handleFromClass)
+			PUT("/foo/", :: handleFromClass)
+			PATCH("/foo/") {
+				ok().buildAndAwait()
+			}
+			"/foo/"  { handleFromClass(it) }
+		}
+		"/content".nest {
+			accept(APPLICATION_ATOM_XML, ::handle)
+			contentType(APPLICATION_OCTET_STREAM, ::handle)
+		}
+		method(PATCH, ::handle)
+		headers { it.accept().contains(APPLICATION_JSON) }.nest {
+			GET("/api/foo/", ::handle)
+		}
+		headers({ it.header("bar").isNotEmpty() }, ::handle)
+		resources("/org/springframework/web/reactive/function/**",
+				ClassPathResource("/org/springframework/web/reactive/function/response.txt"))
+		resources {
+			if (it.path() == "/response.txt") {
+				ClassPathResource("/org/springframework/web/reactive/function/response.txt")
+			}
+			else {
+				null
+			}
+		}
+		path("/baz", ::handle)
+		GET("/rendering") { RenderingResponse.create("index").buildAndAwait() }
+	}
+}
+
+@Suppress("UNUSED_PARAMETER")
+private suspend fun handleFromClass(req: ServerRequest) = ServerResponse.ok().buildAndAwait()
+
+@Suppress("UNUSED_PARAMETER")
+private suspend fun handle(req: ServerRequest) = ServerResponse.ok().buildAndAwait()

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

@@ -0,0 +1,41 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.reactive.function.server
+
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.junit.Test
+import reactor.core.publisher.Mono
+
+class RenderingResponseExtensionsTests {
+
+	@Test
+	fun buildAndAwait() {
+		val builder = mockk<RenderingResponse.Builder>()
+		val response = mockk<RenderingResponse>()
+		every { builder.build() } returns Mono.just(response)
+		runBlocking {
+			builder.buildAndAwait()
+		}
+		verify {
+			builder.build()
+		}
+	}
+
+}

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

@@ -27,7 +27,7 @@ import reactor.test.StepVerifier
 import java.net.URI
 
 /**
- * Tests for [RouterFunction] Kotlin DSL.
+ * Tests for [RouterFunctionDsl].
  *
  * @author Sebastien Deleuze
  */
@@ -161,7 +161,7 @@ class RouterFunctionDslTests {
 }
 
 @Suppress("UNUSED_PARAMETER")
-fun handleFromClass(req: ServerRequest) = ServerResponse.ok().build()
+private fun handleFromClass(req: ServerRequest) = ServerResponse.ok().build()
 
 @Suppress("UNUSED_PARAMETER")
-fun handle(req: ServerRequest) = ServerResponse.ok().build()
+private fun handle(req: ServerRequest) = ServerResponse.ok().build()

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

@@ -16,13 +16,20 @@
 
 package org.springframework.web.reactive.function.client
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
 import org.junit.Test
 import org.springframework.core.ParameterizedTypeReference
-import org.springframework.web.reactive.function.server.ServerRequest
-import org.springframework.web.reactive.function.server.bodyToFlux
-import org.springframework.web.reactive.function.server.bodyToMono
+import org.springframework.http.codec.multipart.Part
+import org.springframework.util.MultiValueMap
+import org.springframework.web.reactive.function.server.*
+import org.springframework.web.server.WebSession
+import reactor.core.publisher.Mono
+import java.security.Principal
 
 /**
  * Mock object based tests for [ServerRequest] Kotlin extensions.
@@ -45,5 +52,58 @@ class ServerRequestExtensionsTests {
 		verify { request.bodyToFlux(object : ParameterizedTypeReference<List<Foo>>() {}) }
 	}
 
+	@Test
+	fun awaitBody() {
+		every { request.bodyToMono<String>() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", request.awaitBody<String>())
+		}
+	}
+
+	@Test
+	fun awaitBodyNull() {
+		every { request.bodyToMono<String>() } returns Mono.empty()
+		runBlocking {
+			assertNull(request.awaitBody<String>())
+		}
+	}
+
+	@Test
+	fun awaitFormData() {
+		val map = mockk<MultiValueMap<String, String>>()
+		every { request.formData() } returns Mono.just(map)
+		runBlocking {
+			assertEquals(map, request.awaitFormData())
+		}
+	}
+
+	@Test
+	fun awaitMultipartData() {
+		val map = mockk<MultiValueMap<String, Part>>()
+		every { request.multipartData() } returns Mono.just(map)
+		runBlocking {
+			assertEquals(map, request.awaitMultipartData())
+		}
+	}
+
+	@Test
+	fun awaitPrincipal() {
+		val principal = mockk<Principal>()
+		every { request.principal() } returns Mono.just(principal)
+		runBlocking {
+			assertEquals(principal, request.awaitPrincipal())
+		}
+	}
+
+	@Test
+	fun awaitSession() {
+		val session = mockk<WebSession>()
+		every { request.session() } returns Mono.just(session)
+		runBlocking {
+			assertEquals(session, request.awaitSession())
+		}
+	}
+
+
 	class Foo
 }

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

@@ -16,12 +16,16 @@
 
 package org.springframework.web.reactive.function.server
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertEquals
 import org.junit.Test
 import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
 import org.springframework.http.MediaType.*
+import reactor.core.publisher.Mono
 
 /**
  * Mock object based tests for [ServerResponse] Kotlin extensions
@@ -30,7 +34,7 @@ import org.springframework.http.MediaType.*
  */
 class ServerResponseExtensionsTests {
 
-	val bodyBuilder = mockk<ServerResponse.BodyBuilder>(relaxed = true)
+	private val bodyBuilder = mockk<ServerResponse.BodyBuilder>(relaxed = true)
 
 
 	@Test
@@ -65,5 +69,53 @@ class ServerResponseExtensionsTests {
 		verify { bodyBuilder.contentType(TEXT_HTML) }
 	}
 
+	@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
+	fun `renderAndAwait with a vararg parameter`() {
+		val response = mockk<ServerResponse>()
+		every { bodyBuilder.render("foo", any(), any()) } returns Mono.just(response)
+		runBlocking {
+			bodyBuilder.renderAndAwait("foo", "bar", "baz")
+		}
+		verify {
+			bodyBuilder.render("foo", any(), any())
+		}
+	}
+
+	@Test
+	fun `renderAndAwait with a Map parameter`() {
+		val response = mockk<ServerResponse>()
+		val map = mockk<Map<String, *>>()
+		every { bodyBuilder.render("foo", map) } returns Mono.just(response)
+		runBlocking {
+			bodyBuilder.renderAndAwait("foo", map)
+		}
+		verify {
+			bodyBuilder.render("foo", map)
+		}
+	}
+
 	class Foo
 }