DATACASS-632 - Add ReactiveFluentCassandraOperations Coroutines extensions.

This commit introduces Coroutines support for ReactiveFluentCassandraOperations 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 is fixed.

spring-data-cassandra/pom.xml

@@ -182,22 +182,34 @@
 			<artifactId>kotlin-stdlib</artifactId>
 			<optional>true</optional>
 		</dependency>
+
 		<dependency>
 			<groupId>org.jetbrains.kotlin</groupId>
 			<artifactId>kotlin-reflect</artifactId>
 			<optional>true</optional>
 		</dependency>
+
 		<dependency>
-			<groupId>org.jetbrains.kotlin</groupId>
-			<artifactId>kotlin-test-junit</artifactId>
-			<scope>test</scope>
+			<groupId>org.jetbrains.kotlinx</groupId>
+			<artifactId>kotlinx-coroutines-core</artifactId>
+			<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>
 			<groupId>io.mockk</groupId>
 			<artifactId>mockk</artifactId>
 			<version>${mockk}</version>
 			<scope>test</scope>
 		</dependency>
+
 	</dependencies>
 
 </project>

spring-data-cassandra/src/main/kotlin/org/springframework/data/cassandra/core/ReactiveDeleteOperationExtensions.kt

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.cassandra.core
 
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -35,3 +36,12 @@ fun <T : Any> ReactiveDeleteOperation.delete(entityClass: KClass<T>): ReactiveDe
  */
 inline fun <reified T : Any> ReactiveDeleteOperation.delete(): ReactiveDeleteOperation.ReactiveDelete =
 		delete(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveDeleteOperation.TerminatingDelete.all].
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+suspend fun ReactiveDeleteOperation.TerminatingDelete.allAndAwait(): WriteResult =
+		all().awaitSingle()

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.cassandra.core
 
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -35,3 +36,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 Mark Paluch
+ * @since 2.2
+ */
+suspend inline fun <reified T : Any> ReactiveInsertOperation.TerminatingInsert<T>.oneAndAwait(o: T): EntityWriteResult<T> =
+		one(o).awaitSingle()

spring-data-cassandra/src/main/kotlin/org/springframework/data/cassandra/core/ReactiveSelectOperationExtensions.kt

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.cassandra.core
 
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+import kotlinx.coroutines.reactive.awaitSingle
 import kotlin.reflect.KClass
 
 /**
@@ -47,3 +49,39 @@ fun <T : Any> ReactiveSelectOperation.SelectWithProjection<*>.asType(resultType:
  */
 inline fun <reified T : Any> ReactiveSelectOperation.SelectWithProjection<*>.asType(): ReactiveSelectOperation.SelectWithQuery<T> =
 		`as`(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveSelectOperation.TerminatingSelect.one].
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+suspend inline fun <reified T : Any> ReactiveSelectOperation.TerminatingSelect<T>.awaitOne(): T? =
+		one().awaitFirstOrNull()
+
+/**
+ * Coroutines variant of [ReactiveSelectOperation.TerminatingSelect.first].
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+suspend inline fun <reified T : Any> ReactiveSelectOperation.TerminatingSelect<T>.awaitFirst(): T? =
+		first().awaitFirstOrNull()
+
+/**
+ * Coroutines variant of [ReactiveSelectOperation.TerminatingSelect.count].
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveSelectOperation.TerminatingSelect<T>.awaitCount(): Long =
+		count().awaitSingle()
+
+/**
+ * Coroutines variant of [ReactiveSelectOperation.TerminatingSelect.exists].
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+suspend fun <T : Any> ReactiveSelectOperation.TerminatingSelect<T>.awaitExists(): Boolean =
+		exists().awaitSingle()

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.cassandra.core
 
+import kotlinx.coroutines.reactive.awaitSingle
+import org.springframework.data.cassandra.core.query.Update
 import kotlin.reflect.KClass
 
 /**
@@ -35,3 +37,11 @@ fun <T : Any> ReactiveUpdateOperation.update(entityClass: KClass<T>): ReactiveUp
  */
 inline fun <reified T : Any> ReactiveUpdateOperation.update(): ReactiveUpdateOperation.ReactiveUpdate =
 		update(T::class.java)
+
+/**
+ * Coroutines variant of [ReactiveUpdateOperation.TerminatingUpdate.apply].
+ *
+ * @author MarkPaluch
+ * @since 2.2
+ */
+suspend fun ReactiveUpdateOperation.TerminatingUpdate.applyAndAwait(update: Update): WriteResult = apply(update).awaitSingle()

spring-data-cassandra/src/test/kotlin/org/springframework/data/cassandra/core/ReactiveDeleteOperationExtensionsUnitTests.kt

@@ -15,10 +15,14 @@
  */
 package org.springframework.data.cassandra.core
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions
 import org.junit.Test
 import org.springframework.data.cassandra.domain.Person
+import reactor.core.publisher.Mono
 
 /**
  * Unit tests for [ReactiveDeleteOperationExtensions].
@@ -42,4 +46,20 @@ class ReactiveDeleteOperationExtensionsUnitTests {
 		operations.delete<Person>()
 		verify { operations.delete(Person::class.java) }
 	}
+
+	@Test // DATACASS-632
+	fun allAndAwait() {
+
+		val delete = mockk<ReactiveDeleteOperation.TerminatingDelete>()
+		val result = mockk<WriteResult>()
+		every { delete.all() } returns Mono.just(result)
+
+		runBlocking {
+			Assertions.assertThat(delete.allAndAwait()).isEqualTo(result)
+		}
+
+		verify {
+			delete.all()
+		}
+	}
 }

spring-data-cassandra/src/test/kotlin/org/springframework/data/cassandra/core/ReactiveInsertOperationExtensionsUnitTests.kt

@@ -15,11 +15,14 @@
  */
 package org.springframework.data.cassandra.core
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
-import org.junit.Ignore
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions
 import org.junit.Test
 import org.springframework.data.cassandra.domain.Person
+import reactor.core.publisher.Mono
 
 /**
  * Unit tests for [ReactiveInsertOperationExtensions].
@@ -43,4 +46,20 @@ class ReactiveInsertOperationExtensionsUnitTests {
 		operations.insert<Person>()
 		verify { operations.insert(Person::class.java) }
 	}
+
+	@Test // DATACASS-632
+	fun oneAndAwait() {
+
+		val insert = mockk<ReactiveInsertOperation.TerminatingInsert<String>>()
+		val result = mockk<EntityWriteResult<String>>()
+		every { insert.one("foo") } returns Mono.just(result)
+
+		runBlocking {
+			Assertions.assertThat(insert.oneAndAwait("foo")).isEqualTo(result)
+		}
+
+		verify {
+			insert.one("foo")
+		}
+	}
 }

spring-data-cassandra/src/test/kotlin/org/springframework/data/cassandra/core/ReactiveSelectOperationExtensionsUnitTests.kt

@@ -15,11 +15,15 @@
  */
 package org.springframework.data.cassandra.core
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions
 import org.junit.Test
 import org.springframework.data.cassandra.domain.Person
 import org.springframework.data.cassandra.domain.User
+import reactor.core.publisher.Mono
 
 /**
  * Unit tests for [ReactiveSelectOperationExtensions].
@@ -59,4 +63,64 @@ class ReactiveSelectOperationExtensionsUnitTests {
 		operationWithProjection.asType<User>();
 		verify { operationWithProjection.`as`(User::class.java) }
 	}
+
+	@Test // DATACASS-632
+	fun terminatingFindAwaitOne() {
+
+		val find = mockk<ReactiveSelectOperation.TerminatingSelect<String>>()
+		every { find.one() } returns Mono.just("foo")
+
+		runBlocking {
+			Assertions.assertThat(find.awaitOne()).isEqualTo("foo")
+		}
+
+		verify {
+			find.one()
+		}
+	}
+
+	@Test // DATACASS-632
+	fun terminatingFindAwaitFirst() {
+
+		val find = mockk<ReactiveSelectOperation.TerminatingSelect<String>>()
+		every { find.first() } returns Mono.just("foo")
+
+		runBlocking {
+			Assertions.assertThat(find.awaitFirst()).isEqualTo("foo")
+		}
+
+		verify {
+			find.first()
+		}
+	}
+
+	@Test // DATACASS-632
+	fun terminatingFindAwaitCount() {
+
+		val find = mockk<ReactiveSelectOperation.TerminatingSelect<String>>()
+		every { find.count() } returns Mono.just(1)
+
+		runBlocking {
+			Assertions.assertThat(find.awaitCount()).isEqualTo(1)
+		}
+
+		verify {
+			find.count()
+		}
+	}
+
+	@Test // DATACASS-632
+	fun terminatingFindAwaitExists() {
+
+		val find = mockk<ReactiveSelectOperation.TerminatingSelect<String>>()
+		every { find.exists() } returns Mono.just(true)
+
+		runBlocking {
+			Assertions.assertThat(find.awaitExists()).isTrue()
+		}
+
+		verify {
+			find.exists()
+		}
+	}
 }

spring-data-cassandra/src/test/kotlin/org/springframework/data/cassandra/core/ReactiveUpdateOperationExtensionsUnitTests.kt

@@ -15,10 +15,15 @@
  */
 package org.springframework.data.cassandra.core
 
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions
 import org.junit.Test
+import org.springframework.data.cassandra.core.query.Update
 import org.springframework.data.cassandra.domain.Person
+import reactor.core.publisher.Mono
 
 /**
  * Unit tests for [ReactiveUpdateOperationExtensions].
@@ -42,4 +47,21 @@ class ReactiveUpdateOperationExtensionsUnitTests {
 		operations.update<Person>()
 		verify { operations.update(Person::class.java) }
 	}
+
+	@Test // DATACASS-632
+	fun applyAndAwait() {
+
+		val update = mockk<ReactiveUpdateOperation.TerminatingUpdate>()
+		val result = mockk<WriteResult>()
+		val updateObj = mockk<Update>();
+		every { update.apply(updateObj) } returns Mono.just(result)
+
+		runBlocking {
+			Assertions.assertThat(update.applyAndAwait(updateObj)).isEqualTo(result)
+		}
+
+		verify {
+			update.apply(updateObj)
+		}
+	}
 }

src/main/asciidoc/new-features.adoc

@@ -7,6 +7,7 @@ This chapter summarizes changes and new features for each release.
 == What's new in Spring Data for Apache Cassandra 2.2
 * Read-only properties annotated with `@ReadOnlyProperty` to exclude non-writable properties from entity-bound `INSERT` and `UPDATE` operations.
 * Support for derived `Between` queries.
+* Kotlin Coroutine extensions for `ReactiveFluentCassandraOperations`.
 
 [[new-features.2-1-0]]
 == What's new in Spring Data for Apache Cassandra 2.1