DATAMONGO-2209 - Add ReactiveFluentMongoOperations Coroutines extensions.

This commit introduces Coroutines support for
ReactiveFluentMongoOperations API via Kotlin extensions that provide
suspendable functions prefixed by `await` or suffixed by `AndAwait` for
Mono based APIs.

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

Original pull request: #649.

spring-data-mongodb/pom.xml

@@ -272,9 +272,17 @@
 			<optional>true</optional>
 		</dependency>
 		<dependency>
-			<groupId>org.jetbrains.kotlin</groupId>
+			<groupId>org.jetbrains.kotlinx</groupId>
-			<artifactId>kotlin-test</artifactId>
+			<artifactId>kotlinx-coroutines-core</artifactId>
-			<scope>test</scope>
+			<version>${kotlin-coroutines}</version>
+			<optional>true</optional>
+		</dependency>
+
+		<dependency>
+			<groupId>org.jetbrains.kotlinx</groupId>
+			<artifactId>kotlinx-coroutines-reactor</artifactId>
+			<version>${kotlin-coroutines}</version>
+			<optional>true</optional>
 		</dependency>
 
 		<dependency>

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensions.kt

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mongodb.core
 
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -70,3 +72,39 @@ fun <T : Any> ReactiveFindOperation.DistinctWithProjection.asType(resultType: KC
  */
 inline fun <reified T : Any> ReactiveFindOperation.DistinctWithProjection.asType(): ReactiveFindOperation.DistinctWithProjection =
 		`as`(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveFindOperation.TerminatingFind.one].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend inline fun <reified T: Any> ReactiveFindOperation.TerminatingFind<T>.awaitOne(): T? =
+		one().awaitFirstOrNull()
+
+/**
+ * Coroutines variant of [ReactiveFindOperation.TerminatingFind.first].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend inline fun <reified T: Any> ReactiveFindOperation.TerminatingFind<T>.awaitFirst(): T? =
+		first().awaitFirstOrNull()
+
+/**
+ * Coroutines variant of [ReactiveFindOperation.TerminatingFind.count].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveFindOperation.TerminatingFind<T>.awaitCount(): Long =
+		count().awaitSingle()
+
+/**
+ * Coroutines variant of [ReactiveFindOperation.TerminatingFind.exists].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveFindOperation.TerminatingFind<T>.awaitExists(): Boolean =
+		exists().awaitSingle()

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveInsertOperationExtensions.kt

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.mongodb.core
 
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -34,3 +35,12 @@ fun <T : Any> ReactiveInsertOperation.insert(entityClass: KClass<T>): ReactiveIn
  */
 inline fun <reified T : Any> ReactiveInsertOperation.insert(): ReactiveInsertOperation.ReactiveInsert<T> =
 		insert(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveInsertOperation.TerminatingInsert.one].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend inline fun <reified T: Any> ReactiveInsertOperation.TerminatingInsert<T>.oneAndAwait(o: T): T =
+		one(o).awaitSingle()

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveRemoveOperationExtensions.kt

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mongodb.core
 
+import com.mongodb.client.result.DeleteResult
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -34,3 +36,12 @@ fun <T : Any> ReactiveRemoveOperation.remove(entityClass: KClass<T>): ReactiveRe
  */
 inline fun <reified T : Any> ReactiveRemoveOperation.remove(): ReactiveRemoveOperation.ReactiveRemove<T> =
 		remove(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveRemoveOperation.TerminatingRemove.all].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveRemoveOperation.TerminatingRemove<T>.allAndAwait(): DeleteResult =
+		all().awaitSingle()

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveUpdateOperationExtensions.kt

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.mongodb.core
 
+import com.mongodb.client.result.UpdateResult
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -34,3 +37,57 @@ fun <T : Any> ReactiveUpdateOperation.update(entityClass: KClass<T>): ReactiveUp
  */
 inline fun <reified T : Any> ReactiveUpdateOperation.update(): ReactiveUpdateOperation.ReactiveUpdate<T> =
 		update(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveUpdateOperation.TerminatingFindAndModify.findModifyAndAwait].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveUpdateOperation.TerminatingFindAndModify<T>.findModifyAndAwait(): T? =
+		findAndModify().awaitFirstOrNull()
+
+
+/**
+ * Coroutines variant of [ReactiveUpdateOperation.TerminatingFindAndReplace.findAndReplace].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveUpdateOperation.TerminatingFindAndReplace<T>.findReplaceAndAwait(): T? =
+		findAndReplace().awaitFirstOrNull()
+
+/**
+ * Coroutines variant of [ReactiveUpdateOperation.TerminatingUpdate.all].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveUpdateOperation.TerminatingUpdate<T>.allAndAwait(): UpdateResult =
+		all().awaitSingle()
+
+/**
+ * Coroutines variant of [ReactiveUpdateOperation.TerminatingUpdate.first].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveUpdateOperation.TerminatingUpdate<T>.firstAndAwait(): UpdateResult =
+		first().awaitSingle()
+
+/**
+ * Coroutines variant of [ReactiveUpdateOperation.TerminatingUpdate.upsert].
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveUpdateOperation.TerminatingUpdate<T>.upsertAndAwait(): UpdateResult = upsert().awaitSingle()
+
+/**
+ * Extension for [ReactiveUpdateOperation.FindAndReplaceWithProjection.as] leveraging reified type parameters.
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+inline fun <reified T : Any> ReactiveUpdateOperation.FindAndReplaceWithProjection<T>.asType(): ReactiveUpdateOperation.FindAndReplaceWithOptions<T> =
+		`as`(T::class.java)

spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensionsTests.kt

@@ -16,9 +16,13 @@
 package org.springframework.data.mongodb.core
 
 import example.first.First
+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
 
 /**
  * @author Mark Paluch
@@ -73,4 +77,52 @@ class ReactiveFindOperationExtensionsTests {
 		distinctWithProjection.asType<User>()
 		verify { distinctWithProjection.`as`(User::class.java) }
 	}
+
+	@Test
+	fun terminatingFindAwaitOne() {
+		val find = mockk<ReactiveFindOperation.TerminatingFind<String>>()
+		every { find.one() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", find.awaitOne())
+		}
+		verify {
+			find.one()
+		}
+	}
+
+	@Test
+	fun terminatingFindAwaitFirst() {
+		val find = mockk<ReactiveFindOperation.TerminatingFind<String>>()
+		every { find.first() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", find.awaitFirst())
+		}
+		verify {
+			find.first()
+		}
+	}
+
+	@Test
+	fun terminatingFindAwaitCount() {
+		val find = mockk<ReactiveFindOperation.TerminatingFind<String>>()
+		every { find.count() } returns Mono.just(1)
+		runBlocking {
+			assertEquals(1, find.awaitCount())
+		}
+		verify {
+			find.count()
+		}
+	}
+
+	@Test
+	fun terminatingFindAwaitExists() {
+		val find = mockk<ReactiveFindOperation.TerminatingFind<String>>()
+		every { find.exists() } returns Mono.just(true)
+		runBlocking {
+			assertEquals(true, find.awaitExists())
+		}
+		verify {
+			find.exists()
+		}
+	}
 }

spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveInsertOperationExtensionsTests.kt

@@ -16,9 +16,13 @@
 package org.springframework.data.mongodb.core
 
 import example.first.First
+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
 
 /**
  * @author Mark Paluch
@@ -41,4 +45,16 @@ class ReactiveInsertOperationExtensionsTests {
 		operation.insert<First>()
 		verify { operation.insert(First::class.java) }
 	}
+
+	@Test
+	fun terminatingFindAwaitOne() {
+		val find = mockk<ReactiveInsertOperation.TerminatingInsert<String>>()
+		every { find.one("foo") } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", find.oneAndAwait("foo"))
+		}
+		verify {
+			find.one("foo")
+		}
+	}
 }

spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveRemoveOperationExtensionsTests.kt

@@ -15,10 +15,15 @@
  */
 package org.springframework.data.mongodb.core
 
+import com.mongodb.client.result.DeleteResult
 import example.first.First
+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
 
 /**
  * @author Mark Paluch
@@ -41,4 +46,17 @@ class ReactiveRemoveOperationExtensionsTests {
 		operation.remove<First>()
 		verify { operation.remove(First::class.java) }
 	}
+
+	@Test
+	fun allAndAwait() {
+		val remove = mockk<ReactiveRemoveOperation.TerminatingRemove<String>>()
+		val result = mockk<DeleteResult>()
+		every { remove.all() } returns Mono.just(result)
+		runBlocking {
+			assertEquals(result, remove.allAndAwait())
+		}
+		verify {
+			remove.all()
+		}
+	}
 }

spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveUpdateOperationExtensionsTests.kt

@@ -15,10 +15,15 @@
  */
 package org.springframework.data.mongodb.core
 
+import com.mongodb.client.result.UpdateResult
 import example.first.First
+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
 
 /**
  * Unit tests for `ReactiveExecutableUpdateOperationExtensions.kt`.
@@ -43,4 +48,78 @@ class ReactiveUpdateOperationExtensionsTests {
 		operation.update<First>()
 		verify { operation.update(First::class.java) }
 	}
+
+	@Test
+	fun findModifyAndAwait() {
+		val find = mockk<ReactiveUpdateOperation.TerminatingFindAndModify<String>>()
+		every { find.findAndModify() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", find.findModifyAndAwait())
+		}
+		verify {
+			find.findAndModify()
+		}
+	}
+
+	@Test
+	fun findReplaceAndAwait() {
+		val find = mockk<ReactiveUpdateOperation.TerminatingFindAndReplace<String>>()
+		every { find.findAndReplace() } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", find.findReplaceAndAwait())
+		}
+		verify {
+			find.findAndReplace()
+		}
+	}
+
+	@Test
+	fun allAndAwait() {
+		val update = mockk<ReactiveUpdateOperation.TerminatingUpdate<String>>()
+		val result = mockk<UpdateResult>()
+		every { update.all() } returns Mono.just(result)
+		runBlocking {
+			assertEquals(result, update.allAndAwait())
+		}
+		verify {
+			update.all()
+		}
+	}
+
+	@Test
+	fun firstAndAwait() {
+		val update = mockk<ReactiveUpdateOperation.TerminatingUpdate<String>>()
+		val result = mockk<UpdateResult>()
+		every { update.first() } returns Mono.just(result)
+		runBlocking {
+			assertEquals(result, update.firstAndAwait())
+		}
+		verify {
+			update.first()
+		}
+	}
+
+	@Test
+	fun upsertAndAwait() {
+		val update = mockk<ReactiveUpdateOperation.TerminatingUpdate<String>>()
+		val result = mockk<UpdateResult>()
+		every { update.upsert() } returns Mono.just(result)
+		runBlocking {
+			assertEquals(result, update.upsertAndAwait())
+		}
+		verify {
+			update.upsert()
+		}
+	}
+
+	@Test
+	fun findAndReplaceWithProjectionAsType() {
+		val update = mockk<ReactiveUpdateOperation.FindAndReplaceWithProjection<String>>()
+		val result = mockk<ReactiveUpdateOperation.FindAndReplaceWithOptions<String>>()
+		every { update.`as`(String::class.java) } returns result
+		assertEquals(result, update.asType<String>())
+		verify {
+			update.`as`(String::class.java)
+		}
+	}
 }