DATACASS-648 - Add a Flow extension to ReactiveSelectOperation.

Original pull request: #159.

Author: sdeleuze

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

@@ -15,8 +15,11 @@
  */
 package org.springframework.data.cassandra.core
 
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.reactive.awaitFirstOrNull
 import kotlinx.coroutines.reactive.awaitSingle
+import kotlinx.coroutines.reactive.flow.asFlow
 import kotlin.reflect.KClass
 
 /**
@@ -107,3 +110,16 @@ suspend fun <T : Any> ReactiveSelectOperation.TerminatingSelect<T>.awaitCount():
  */
 suspend fun <T : Any> ReactiveSelectOperation.TerminatingSelect<T>.awaitExists(): Boolean =
 		exists().awaitSingle()
+
+/**
+ * Coroutines [Flow] variant of [ReactiveSelectOperation.TerminatingSelect.all].
+ *
+ * Backpressure is controlled by [batchSize] parameter that controls the size of in-flight elements
+ * and [org.reactivestreams.Subscription.request] size.
+ *
+ * @author Sebastien Deleuze
+ * @since 2.2
+ */
+@FlowPreview
+fun <T : Any> ReactiveSelectOperation.TerminatingSelect<T>.allAsFlow(batchSize: Int = 1): Flow<T> =
+		all().asFlow(batchSize)

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

@@ -18,11 +18,14 @@ package org.springframework.data.cassandra.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.flow.toList
 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.Flux
 import reactor.core.publisher.Mono
 
 /**
@@ -213,4 +216,20 @@ class ReactiveSelectOperationExtensionsUnitTests {
 			find.exists()
 		}
 	}
+
+	@Test
+	@FlowPreview
+	fun terminatingFindAllAsFlow() {
+
+		val spec = mockk<ReactiveSelectOperation.TerminatingSelect<String>>()
+		every { spec.all() } returns Flux.just("foo", "bar", "baz")
+
+		runBlocking {
+			Assertions.assertThat(spec.allAsFlow().toList()).contains("foo", "bar", "baz")
+		}
+
+		verify {
+			spec.all()
+		}
+	}
 }