#63 - Add DatabaseClient Coroutines extensions.

This commit introduces Coroutines support for `DatabaseClient`
functional API via Kotlin extensions that provide suspendable
functions prefixed by `await` for `Mono` based APIs.

Extensions for `Flux` will be added when Kotlin/kotlinx.coroutines#254
will be fixed.

It also provides `asType<Foo>()` extensions useful for Reactive API
as well.

Original pull request: #63.

pom.xml

@@ -35,6 +35,7 @@
 		<r2dbc-releasetrain.version>Arabba-M7</r2dbc-releasetrain.version>
 		<reactive-streams.version>1.0.1</reactive-streams.version>
 		<testcontainers.version>1.10.1</testcontainers.version>
+		<coroutines.version>1.1.1</coroutines.version>
 
 	</properties>
 
@@ -126,6 +127,32 @@
 			<artifactId>reactor-core</artifactId>
 		</dependency>
 
+		<!-- Kotlin extension -->
+		<dependency>
+			<groupId>org.jetbrains.kotlin</groupId>
+			<artifactId>kotlin-stdlib</artifactId>
+			<version>${kotlin}</version>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.jetbrains.kotlin</groupId>
+			<artifactId>kotlin-reflect</artifactId>
+			<version>${kotlin}</version>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.jetbrains.kotlinx</groupId>
+			<artifactId>kotlinx-coroutines-core</artifactId>
+			<version>${coroutines.version}</version>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.jetbrains.kotlinx</groupId>
+			<artifactId>kotlinx-coroutines-reactor</artifactId>
+			<version>${coroutines.version}</version>
+			<optional>true</optional>
+		</dependency>
+
 		<dependency>
 			<groupId>org.assertj</groupId>
 			<artifactId>assertj-core</artifactId>
@@ -198,6 +225,13 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>io.mockk</groupId>
+			<artifactId>mockk</artifactId>
+			<version>1.9.1</version>
+			<scope>test</scope>
+		</dependency>
+
 	</dependencies>
 
 	<build>

src/main/kotlin/org/springframework/data/r2dbc/function/DatabaseClientExtensions.kt

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2018-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.data.r2dbc.function
+
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+
+/**
+ * Coroutines variant of [DatabaseClient.GenericExecuteSpec.then].
+ *
+ * @author Sebastien Deleuze
+ */
+suspend fun DatabaseClient.GenericExecuteSpec.await() {
+    then().awaitFirstOrNull()
+}
+
+/**
+ * Extension for [DatabaseClient.GenericExecuteSpec.as] providing a
+ * `asType<Foo>()` variant.
+ *
+ * @author Sebastien Deleuze
+ */
+inline fun <reified T : Any> DatabaseClient.GenericExecuteSpec.asType(): DatabaseClient.TypedExecuteSpec<T>
+        = `as`(T::class.java)
+
+/**
+ * Extension for [DatabaseClient.GenericSelectSpec.as] providing a
+ * `asType<Foo>()` variant.
+ *
+ * @author Sebastien Deleuze
+ */
+inline fun <reified T : Any> DatabaseClient.GenericSelectSpec.asType(): DatabaseClient.TypedSelectSpec<T>
+        = `as`(T::class.java)
+
+/**
+ * Coroutines variant of [DatabaseClient.TypedExecuteSpec.then].
+ *
+ * @author Sebastien Deleuze
+ */
+suspend fun <T> DatabaseClient.TypedExecuteSpec<T>.await() {
+    then().awaitFirstOrNull()
+}
+
+/**
+ * Extension for [DatabaseClient.TypedExecuteSpec.as] providing a
+ * `asType<Foo>()` variant.
+ *
+ * @author Sebastien Deleuze
+ */
+inline fun <reified T : Any> DatabaseClient.TypedExecuteSpec<T>.asType(): DatabaseClient.TypedExecuteSpec<T>
+        = `as`(T::class.java)
+
+/**
+ * Coroutines variant of [DatabaseClient.InsertSpec.then].
+ *
+ * @author Sebastien Deleuze
+ */
+suspend fun <T> DatabaseClient.InsertSpec<T>.await() {
+    then().awaitFirstOrNull()
+}
+
+/**
+ * Extension for [DatabaseClient.InsertIntoSpec.into] providing a
+ * `into<Foo>()` variant.
+ *
+ * @author Sebastien Deleuze
+ */
+inline fun <reified T : Any> DatabaseClient.InsertIntoSpec.into(): DatabaseClient.TypedInsertSpec<T>
+        = into(T::class.java)
+

src/main/kotlin/org/springframework/data/r2dbc/function/RowsFetchSpecExtensions.kt

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2018-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.data.r2dbc.function
+
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+
+/**
+ * Coroutines variant of [RowsFetchSpec.one].
+ *
+ * @author Sebastien Deleuze
+ */
+suspend fun <T> RowsFetchSpec<T>.awaitOne(): T?
+        = one().awaitFirstOrNull()
+
+/**
+ * Coroutines variant of [RowsFetchSpec.first].
+ *
+ * @author Sebastien Deleuze
+ */
+suspend fun <T> RowsFetchSpec<T>.awaitFirst(): T?
+        = first().awaitFirstOrNull()
+
+// TODO Coroutines variant of [RowsFetchSpec.all], depends on [kotlinx.coroutines#254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
+// suspend fun <T> RowsFetchSpec<T>.awaitAll() = all()...

src/test/kotlin/org/springframework/data/r2dbc/function/DatabaseClientExtensionsTests.kt

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2018-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.data.r2dbc.function
+
+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 reactor.core.publisher.Mono
+
+class DatabaseClientExtensionsTests {
+
+    @Test
+    fun genericExecuteSpecAwait() {
+        val spec = mockk<DatabaseClient.GenericExecuteSpec>()
+        every { spec.then() } returns Mono.empty()
+        runBlocking {
+            spec.await()
+        }
+        verify {
+            spec.then()
+        }
+    }
+
+    @Test
+    fun genericExecuteSpecAsType() {
+        val genericSpec = mockk<DatabaseClient.GenericExecuteSpec>()
+        val typedSpec: DatabaseClient.TypedExecuteSpec<String> = mockk()
+        every { genericSpec.`as`(String::class.java) } returns typedSpec
+        runBlocking {
+            assertEquals(typedSpec, genericSpec.asType<String>())
+        }
+        verify {
+            genericSpec.`as`(String::class.java)
+        }
+    }
+
+    @Test
+    fun genericSelectSpecAsType() {
+        val genericSpec = mockk<DatabaseClient.GenericSelectSpec>()
+        val typedSpec: DatabaseClient.TypedSelectSpec<String> = mockk()
+        every { genericSpec.`as`(String::class.java) } returns typedSpec
+        runBlocking {
+            assertEquals(typedSpec, genericSpec.asType<String>())
+        }
+        verify {
+            genericSpec.`as`(String::class.java)
+        }
+    }
+
+    @Test
+    fun typedExecuteSpecAwait() {
+        val spec = mockk<DatabaseClient.TypedExecuteSpec<String>>()
+        every { spec.then() } returns Mono.empty()
+        runBlocking {
+            spec.await()
+        }
+        verify {
+            spec.then()
+        }
+    }
+
+    @Test
+    fun typedExecuteSpecAsType() {
+        val spec: DatabaseClient.TypedExecuteSpec<String> = mockk()
+        every { spec.`as`(String::class.java) } returns spec
+        runBlocking {
+            assertEquals(spec, spec.asType())
+        }
+        verify {
+            spec.`as`(String::class.java)
+        }
+    }
+
+    @Test
+    fun insertSpecAwait() {
+        val spec = mockk<DatabaseClient.InsertSpec<String>>()
+        every { spec.then() } returns Mono.empty()
+        runBlocking {
+            spec.await()
+        }
+        verify {
+            spec.then()
+        }
+    }
+
+    @Test
+    fun insertIntoSpecInto() {
+        val spec = mockk<DatabaseClient.InsertIntoSpec>()
+        val typedSpec: DatabaseClient.TypedInsertSpec<String> = mockk()
+        every { spec.into(String::class.java) } returns typedSpec
+        runBlocking {
+            assertEquals(typedSpec, spec.into<String>())
+        }
+        verify {
+            spec.into(String::class.java)
+        }
+    }
+}

src/test/kotlin/org/springframework/data/r2dbc/function/RowsFetchSpecExtensionsTests.kt

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2018-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.data.r2dbc.function
+
+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 reactor.core.publisher.Mono
+
+class RowsFetchSpecExtensionsTests {
+
+    @Test
+    fun awaitOne() {
+        val spec = mockk<RowsFetchSpec<String>>()
+        every { spec.one() } returns Mono.just("foo")
+        runBlocking {
+            assertEquals("foo", spec.awaitOne())
+        }
+        verify {
+            spec.one()
+        }
+    }
+
+    @Test
+    fun awaitFirst() {
+        val spec = mockk<RowsFetchSpec<String>>()
+        every { spec.first() } returns Mono.just("foo")
+        runBlocking {
+            assertEquals("foo", spec.awaitFirst())
+        }
+        verify {
+            spec.first()
+        }
+    }
+}