DATACASS-648 - Add a Flow extension to ReactiveSelectOperation

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()
+}
+}
 }