DATAREDIS-1033 - Add Kotlin Flow based extensions.

Original Pull Request: #477
2019-09-10 06:26:30 +02:00
parent 669d3a9c5c
commit 78b895feb4
16 changed files with 1541 additions and 22 deletions
--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensionsUnitTests.kt
@@ -18,17 +18,26 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
+import org.reactivestreams.Publisher
+import org.springframework.data.geo.Circle
 import org.springframework.data.geo.Distance
+import org.springframework.data.geo.GeoResult
 import org.springframework.data.geo.Metrics
 import org.springframework.data.geo.Point
 import org.springframework.data.redis.connection.RedisGeoCommands
+import org.springframework.data.redis.connection.RedisGeoCommands.*
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono

 /**
- * Unit tests for [ReactiveGeoOperationsExtensions].
+ * Unit tests for `ReactiveGeoOperationsExtensions`.
 *
 * @author Mark Paluch
 * @author Christoph Strobl
@@ -54,14 +63,14 @@ class ReactiveGeoOperationsExtensionsUnitTests {
 	fun addGeoLocation() {

 		val operations = mockk<ReactiveGeoOperations<String, String>>()
-		every { operations.add(any(), any<RedisGeoCommands.GeoLocation<String>>()) } returns Mono.just(1)
+		every { operations.add(any(), any<GeoLocation<String>>()) } returns Mono.just(1)

 		runBlocking {
-			assertThat(operations.addAndAwait("foo", RedisGeoCommands.GeoLocation("bar", Point(1.0, 2.0)))).isEqualTo(1)
+			assertThat(operations.addAndAwait("foo", GeoLocation("bar", Point(1.0, 2.0)))).isEqualTo(1)
 		}

 		verify {
-			operations.add("foo", RedisGeoCommands.GeoLocation("bar", Point(1.0, 2.0)))
+			operations.add("foo", GeoLocation("bar", Point(1.0, 2.0)))
 		}
 	}

@@ -84,14 +93,30 @@ class ReactiveGeoOperationsExtensionsUnitTests {
 	fun addGeoLocationList() {

 		val operations = mockk<ReactiveGeoOperations<String, String>>()
-		every { operations.add(any(), any<List<RedisGeoCommands.GeoLocation<String>>>()) } returns Mono.just(1)
+		every { operations.add(any(), any<List<GeoLocation<String>>>()) } returns Mono.just(1)

 		runBlocking {
-			assertThat(operations.addAndAwait("foo", listOf(RedisGeoCommands.GeoLocation("bar", Point(1.0, 2.0))))).isEqualTo(1)
+			assertThat(operations.addAndAwait("foo", listOf(GeoLocation("bar", Point(1.0, 2.0))))).isEqualTo(1)
 		}

 		verify {
-			operations.add("foo", listOf(RedisGeoCommands.GeoLocation("bar", Point(1.0, 2.0))))
+			operations.add("foo", listOf(GeoLocation("bar", Point(1.0, 2.0))))
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun addGeoLocationFlow() {
+		val operations = mockk<ReactiveGeoOperations<String, String>>()
+		every { operations.add(any(), any<Publisher<List<GeoLocation<String>>>>()) } returns Flux.just(1)
+		val flow = flow { emit(listOf(GeoLocation("bar", Point(1.0, 2.0)))) }
+
+		runBlocking {
+			assertThat(operations.add("foo", flow).toList()).contains(1)
+		}
+
+		verify {
+			operations.add("foo", any<Publisher<List<GeoLocation<String>>>>())
 		}
 	}

@@ -246,6 +271,95 @@ class ReactiveGeoOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun radiusAsFlowCircle() {
+		val operations = mockk<ReactiveGeoOperations<String, String>>()
+		val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0))
+		val circle = Circle(1.0, 2.0, 3.0)
+		every { operations.radius(any(), any()) } returns Flux.just(result)
+
+		runBlocking {
+			assertThat(operations.radiusAsFlow("foo", circle).toList()).contains(result)
+		}
+
+		verify {
+			operations.radius("foo", circle)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun radiusAsFlowCircleAndArgs() {
+		val operations = mockk<ReactiveGeoOperations<String, String>>()
+		val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0))
+		val circle = Circle(1.0, 2.0, 3.0)
+		val args = GeoRadiusCommandArgs.newGeoRadiusArgs()
+		every { operations.radius(any(), any(), args) } returns Flux.just(result)
+
+		runBlocking {
+			assertThat(operations.radiusAsFlow("foo", circle, args).toList()).contains(result)
+		}
+
+		verify {
+			operations.radius("foo", circle, args)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun radiusAsFlowMemberAndRadius() {
+		val operations = mockk<ReactiveGeoOperations<String, String>>()
+		val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0))
+
+		every { operations.radius(any(), any(), any<Double>()) } returns Flux.just(result)
+
+		runBlocking {
+			assertThat(operations.radiusAsFlow("foo", "bar", 1.0).toList()).contains(result)
+		}
+
+		verify {
+			operations.radius("foo", "bar", 1.0)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun radiusAsFlowDistance() {
+		val operations = mockk<ReactiveGeoOperations<String, String>>()
+		val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0))
+		val distance = Distance(2.0)
+
+		every { operations.radius(any(), any(), any<Distance>()) } returns Flux.just(result)
+
+		runBlocking {
+			assertThat(operations.radiusAsFlow("foo", "bar", distance).toList()).contains(result)
+		}
+
+		verify {
+			operations.radius("foo", "bar", distance)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun radiusAsFlowDistanceAndArgs() {
+		val operations = mockk<ReactiveGeoOperations<String, String>>()
+		val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0))
+		val distance = Distance(2.0)
+		val args = GeoRadiusCommandArgs.newGeoRadiusArgs()
+
+		every { operations.radius(any(), any(), any(), any()) } returns Flux.just(result)
+
+		runBlocking {
+			assertThat(operations.radiusAsFlow("foo", "bar", distance, args).toList()).contains(result)
+		}
+
+		verify {
+			operations.radius("foo", "bar", distance, args)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun remove() {

--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensionsUnitTests.kt
@@ -18,13 +18,16 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono

 /**
- * Unit tests for [ReactiveHashOperationsExtensions].
+ * Unit tests for `ReactiveHashOperationsExtensions`.
 *
 * @author Mark Paluch
 * @author Christoph Strobl
@@ -106,6 +109,21 @@ class ReactiveHashOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun keys() {
+		val operations = mockk<ReactiveHashOperations<String, String, String>>()
+		every { operations.keys(any()) } returns Flux.just("bar", "baz")
+
+		runBlocking {
+			assertThat(operations.keysAsFlow("foo").toList()).contains("bar", "baz")
+		}
+
+		verify {
+			operations.keys("foo")
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun incrementDouble() {

@@ -181,6 +199,56 @@ class ReactiveHashOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun values() {
+
+		val operations = mockk<ReactiveHashOperations<String, String, String>>()
+		every { operations.values(any()) } returns Flux.just("bar", "baz")
+
+		runBlocking {
+			assertThat(operations.valuesAsFlow("foo").toList()).contains("bar","baz")
+		}
+
+		verify {
+			operations.values("foo")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun entries() {
+
+		val entry = java.util.AbstractMap.SimpleEntry("bar", "baz")
+		val operations = mockk<ReactiveHashOperations<String, String, String>>()
+		every { operations.entries(any()) } returns Flux.just(entry)
+
+		runBlocking {
+			assertThat(operations.entriesAsFlow("foo").toList()).contains(entry)
+		}
+
+		verify {
+			operations.entries("foo")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun scan() {
+
+		val entry = java.util.AbstractMap.SimpleEntry("bar", "baz")
+		val operations = mockk<ReactiveHashOperations<String, String, String>>()
+		every { operations.scan(any(), any()) } returns Flux.just(entry)
+
+		runBlocking {
+			assertThat(operations.scanAsFlow("foo").toList()).contains(entry)
+		}
+
+		verify {
+			operations.scan("foo", ScanOptions.NONE)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun remove() {

--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveHyperLogLogOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveHyperLogLogOperationsExtensionsUnitTests.kt
@@ -24,7 +24,7 @@ import org.junit.Test
 import reactor.core.publisher.Mono

 /**
- * Unit tests for [ReactiveHyperLogLogOperationsExtensions]
+ * Unit tests for `ReactiveHyperLogLogOperationsExtensions`
 *
 * @author Mark Paluch
 */
--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensionsUnitTests.kt
@@ -18,19 +18,38 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.time.Duration

 /**
- * Unit tests for [ReactiveListOperationsExtensions]
+ * Unit tests for `ReactiveListOperationsExtensions`
 *
 * @author Mark Paluch
 */
 class ReactiveListOperationsExtensionsUnitTests {

+	@Test
+	@ExperimentalCoroutinesApi
+	fun range() {
+
+		val operations = mockk<ReactiveListOperations<String, String>>()
+		every { operations.range(any(), any(), any()) } returns Flux.just("foo", "bar")
+
+		runBlocking {
+			assertThat(operations.rangeAsFlow("foo", 2, 3).toList()).contains("foo", "bar")
+		}
+
+		verify {
+			operations.range("foo", 2, 3)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun trim() {

--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt
@@ -18,22 +18,83 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
 import org.springframework.data.redis.connection.DataType
+import org.springframework.data.redis.connection.ReactiveSubscription
+import org.springframework.data.redis.core.script.RedisScript
+import org.springframework.data.redis.listener.ChannelTopic
+import org.springframework.data.redis.serializer.RedisElementReader
+import org.springframework.data.redis.serializer.RedisElementWriter
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.time.Duration
 import java.time.Instant

 /**
- * Unit tests for [ReactiveRedisOperationsExtensions].
+ * Unit tests for `ReactiveRedisOperationsExtensions`.
 *
 * @author Mark Paluch
 * @author Christoph Strobl
 */
 class ReactiveRedisOperationsExtensionsUnitTests {

+	@Test
+	@ExperimentalCoroutinesApi
+	fun `execute with calllback`() {
+
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.execute(any<ReactiveRedisCallback<*>>()) } returns Flux.just("foo")
+
+		runBlocking {
+			assertThat(operations.executeAsFlow { flow { emit("foo")} }.toList()).contains("foo")
+		}
+
+		verify {
+			operations.execute(any<ReactiveRedisCallback<*>>())
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `execute with script`() {
+
+		val script = RedisScript.of<String>("foo")
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.execute(any<RedisScript<*>>(), any(), any()) } returns Flux.just("foo")
+
+		runBlocking {
+			assertThat(operations.executeAsFlow(script).toList()).contains("foo")
+		}
+
+		verify {
+			operations.execute(script, any(), any())
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `execute with script, argsWriter and resultReader`() {
+
+		val script = RedisScript.of<String>("foo")
+		val argsWriter = mockk<RedisElementWriter<Any>>(relaxed = true)
+		val resultReader = mockk<RedisElementReader<String>>(relaxed = true)
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.execute(any<RedisScript<*>>(), any(), any(), any(), any()) } returns Flux.just("foo")
+
+		runBlocking {
+			assertThat(operations.executeAsFlow(script, argsWriter = argsWriter, resultReader = resultReader).toList()).contains("foo")
+		}
+
+		verify {
+			operations.execute(script, any(), any(), argsWriter, resultReader)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun convertAndSend() {

@@ -49,6 +110,59 @@ class ReactiveRedisOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun listenToChannel() {
+
+		val message = ReactiveSubscription.ChannelMessage("a", "b")
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.listenToChannel(any(), any()) } returns Flux.just(message)
+
+		runBlocking {
+			assertThat(operations.listenToChannelAsFlow("foo", "bar").toList()).contains(message)
+		}
+
+		verify {
+			operations.listenToChannel("foo", "bar")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun listenToPattern() {
+
+		val message = ReactiveSubscription.ChannelMessage("a", "b")
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.listenToPattern(any(), any()) } returns Flux.just(message)
+
+		runBlocking {
+			assertThat(operations.listenToPatternAsFlow("foo", "bar").toList()).contains(message)
+		}
+
+		verify {
+			operations.listenToPattern("foo", "bar")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun listenTo() {
+
+		val topic1 = ChannelTopic.of("foo")
+		val topic2 = ChannelTopic.of("bar")
+		val message = ReactiveSubscription.ChannelMessage("a", "b")
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.listenTo(any(), any()) } returns Flux.just(message)
+
+		runBlocking {
+			assertThat(operations.listenToAsFlow(topic1, topic2).toList()).contains(message)
+		}
+
+		verify {
+			operations.listenTo(topic1, topic2)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun hasKey() {

@@ -79,6 +193,39 @@ class ReactiveRedisOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun keys() {
+
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.keys(any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.keysAsFlow("foo").toList()).contains("bar")
+		}
+
+		verify {
+			operations.keys("foo")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun scan() {
+
+		val operations = mockk<ReactiveRedisOperations<String, String>>()
+		every { operations.scan(ScanOptions.NONE) } returns Flux.just("foo")
+
+		runBlocking {
+			assertThat(operations.scanAsFlow().toList()).contains("foo")
+		}
+
+		verify {
+			operations.scan(ScanOptions.NONE)
+		}
+	}
+
+
 	@Test // DATAREDIS-937
 	fun randomKey() {

--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensionsUnitTests.kt
@@ -18,13 +18,16 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono

 /**
- * Unit tests for [ReactiveSetOperationsExtensions].
+ * Unit tests for `ReactiveSetOperationsExtensions`.
 *
 * @author Mark Paluch
 * @author Christoph Strobl
@@ -76,6 +79,22 @@ class ReactiveSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun `pop as Flow`() {
+
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.pop(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.popAsFlow("foo", 1).toList()).contains("bar")
+		}
+
+		verify {
+			operations.pop("foo", 1)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun `pop returning an empty Mono`() {

@@ -136,6 +155,51 @@ class ReactiveSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun intersect() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.intersect("foo", "bar") } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.intersectAsFlow("foo", "bar").toList()).contains("baz")
+		}
+
+		verify {
+			operations.intersect("foo", "bar")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `intersect with key and collection`() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.intersect("foo", listOf("bar")) } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.intersectAsFlow("foo", listOf("bar")).toList()).contains("baz")
+		}
+
+		verify {
+			operations.intersect("foo", listOf("bar"))
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `intersect with collection`() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.intersect(listOf("bar")) } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.intersectAsFlow(listOf("bar")).toList()).contains("baz")
+		}
+
+		verify {
+			operations.intersect(listOf("bar"))
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun intersectAndStore() {

@@ -166,6 +230,51 @@ class ReactiveSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun union() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.union("foo", "bar") } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.unionAsFlow("foo", "bar").toList()).contains("baz")
+		}
+
+		verify {
+			operations.union("foo", "bar")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `union with key and collection`() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.union("foo", listOf("bar")) } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.unionAsFlow("foo", listOf("bar")).toList()).contains("baz")
+		}
+
+		verify {
+			operations.union("foo", listOf("bar"))
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `union with collection`() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.union(listOf("bar")) } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.unionAsFlow(listOf("bar")).toList()).contains("baz")
+		}
+
+		verify {
+			operations.union(listOf("bar"))
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun unionAndStore() {

@@ -196,6 +305,51 @@ class ReactiveSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun difference() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.difference("foo", "bar") } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.differenceAsFlow("foo", "bar").toList()).contains("baz")
+		}
+
+		verify {
+			operations.difference("foo", "bar")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `difference with key and collection`() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.difference("foo", listOf("bar")) } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.differenceAsFlow("foo", listOf("bar")).toList()).contains("baz")
+		}
+
+		verify {
+			operations.difference("foo", listOf("bar"))
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `difference with collection`() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.difference(listOf("bar")) } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.differenceAsFlow(listOf("bar")).toList()).contains("baz")
+		}
+
+		verify {
+			operations.difference(listOf("bar"))
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun differenceAndStore() {

@@ -226,6 +380,37 @@ class ReactiveSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun members() {
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.members("foo") } returns Flux.just("baz")
+
+		runBlocking {
+			assertThat(operations.membersAsFlow("foo").toList()).contains("baz")
+		}
+
+		verify {
+			operations.members("foo")
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun scan() {
+
+		val operations =  mockk<ReactiveSetOperations<String, String>>()
+		every { operations.scan(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.scanAsFlow("foo").toList()).contains("bar")
+		}
+
+		verify {
+			operations.scan("foo", ScanOptions.NONE)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun randomMember() {

@@ -256,6 +441,38 @@ class ReactiveSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun distinctRandomMembers() {
+
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.distinctRandomMembers(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.distinctRandomMembersAsFlow("foo", 1).toList()).contains("bar")
+		}
+
+		verify {
+			operations.distinctRandomMembers("foo", 1)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun randomMembers() {
+
+		val operations = mockk<ReactiveSetOperations<String, String>>()
+		every { operations.randomMembers(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.randomMembersAsFlow("foo", 1).toList()).contains("bar")
+		}
+
+		verify {
+			operations.randomMembers("foo", 1)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun delete() {

--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt
@@ -18,14 +18,21 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
+import org.reactivestreams.Publisher
+import org.springframework.data.domain.Range
+import org.springframework.data.redis.connection.RedisZSetCommands.Limit
 import org.springframework.data.redis.connection.stream.*
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono

 /**
- * Unit tests for [ReactiveStreamOperationsExtensions].
+ * Unit tests for `ReactiveStreamOperationsExtensions`.
 *
 * @author Mark Paluch
 */
@@ -82,11 +89,11 @@ class ReactiveStreamOperationsExtensionsUnitTests {

 		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
 		val record = MapRecord.create("foo", mapOf("a" to "b"))
-		val redordId = RecordId.of("0-0")
-		every { operations.add(record) } returns Mono.just(redordId)
+		val recordId = RecordId.of("0-0")
+		every { operations.add(record) } returns Mono.just(recordId)

 		runBlocking {
-			assertThat(operations.addAndAwait(record)).isEqualTo(redordId)
+			assertThat(operations.addAndAwait(record)).isEqualTo(recordId)
 		}

 		verify {
@@ -94,6 +101,26 @@ class ReactiveStreamOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun `add as Flow`() {
+
+		val map = mapOf("a" to "b")
+		val bodyPublisher = Mono.just(map)
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		val recordId = RecordId.of("0-0")
+		every { operations.add(any(), any<Publisher<Map<String, String>>>()) } returns Flux.just(recordId)
+
+		runBlocking {
+			val bodyFlow = flow { emit(map) }
+			assertThat(operations.add("foo", bodyFlow).toList()).contains(recordId)
+		}
+
+		verify {
+			operations.add("foo", any<Publisher<Map<String, String>>>())
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun addRecord() {

@@ -233,6 +260,230 @@ class ReactiveStreamOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun range() {
+
+		val record = MapRecord.create("foo", mapOf("a" to "b"))
+		val range = Range.just("bar")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.range(any(), any(), any()) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.rangeAsFlow("foo", range).toList()).contains(record)
+		}
+
+		verify {
+			operations.range("foo", range, Limit.unlimited())
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun rangeWithType() {
+
+		val record = ObjectRecord.create("a", "b")
+		val range = Range.just("bar")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.range(any<Class<*>>(), any(), any(), any()) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.rangeWithTypeAsFlow<String, String>("foo", range).toList()).contains(record)
+		}
+
+		verify {
+			operations.range(String::class.java, "foo", range, Limit.unlimited())
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with StreamOffset vararg`() {
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val record = MapRecord.create("foo", mapOf("a" to "b"))
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readAsFlow(offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with options and StreamOffset vararg` () {
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val options = StreamReadOptions.empty()
+		val record = MapRecord.create("foo", mapOf("a" to "b"))
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(options, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readAsFlow(options, offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(options, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with type and StreamOffset vararg`() {
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val record = ObjectRecord.create("a", "b")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(String::class.java, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readWithTypeAsFlow<String, String>(offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(String::class.java, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with type, options and StreamOffset vararg` () {
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val options = StreamReadOptions.empty()
+		val record = ObjectRecord.create("a", "b")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(String::class.java, options, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readWithTypeAsFlow<String, String>(options, offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(String::class.java, options, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with consumer and StreamOffset vararg`() {
+		val consumer = Consumer.from("a", "b")
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val record = MapRecord.create("foo", mapOf("a" to "b"))
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(consumer, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readAsFlow(consumer, offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(consumer, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with consumer, options and StreamOffset vararg`() {
+		val consumer = Consumer.from("a", "b")
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val options = StreamReadOptions.empty()
+		val record = MapRecord.create("foo", mapOf("a" to "b"))
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(consumer, options, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readAsFlow(consumer, options, offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(consumer, options, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with type, consumer and StreamOffset vararg`() {
+		val consumer = Consumer.from("a", "b")
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val record = ObjectRecord.create("a", "b")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(String::class.java, consumer, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readWithTypeAsFlow<String, String>(consumer, offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(String::class.java, consumer, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun `read with type, consumer, options and StreamOffset vararg`() {
+		val consumer = Consumer.from("a", "b")
+		val offset1 = StreamOffset.create("foo", ReadOffset.lastConsumed())
+		val offset2 = StreamOffset.create("bar", ReadOffset.lastConsumed())
+		val options = StreamReadOptions.empty()
+		val record = ObjectRecord.create("a", "b")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.read(String::class.java, consumer, options, offset1, offset2) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.readWithTypeAsFlow<String, String>(consumer, options, offset1, offset2).toList()).contains(record)
+		}
+
+		verify {
+			operations.read(String::class.java, consumer, options, offset1, offset2)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun reverseRange() {
+
+		val record = MapRecord.create("foo", mapOf("a" to "b"))
+		val range = Range.just("bar")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.reverseRange(any(), any(), any()) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.reverseRangeAsFlow("foo", range).toList()).contains(record)
+		}
+
+		verify {
+			operations.reverseRange("foo", range, Limit.unlimited())
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun reverseRangeWithType() {
+
+		val record = ObjectRecord.create("a", "b")
+		val range = Range.just("bar")
+		val operations = mockk<ReactiveStreamOperations<String, String, String>>()
+		every { operations.reverseRange(any<Class<*>>(), any(), any(), any()) } returns Flux.just(record)
+
+		runBlocking {
+			assertThat(operations.reverseRangeWithTypeAsFlow<String, String>("foo", range).toList()).contains(record)
+		}
+
+		verify {
+			operations.reverseRange(String::class.java, "foo", range, Limit.unlimited())
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun trim() {

--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveValueOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveValueOperationsExtensionsUnitTests.kt
@@ -26,7 +26,7 @@ import reactor.core.publisher.Mono
 import java.time.Duration

 /**
- * Unit tests for [ReactiveValueOperationsExtensions].
+ * Unit tests for `ReactiveValueOperationsExtensions`.
 *
 * @author Mark Paluch
 * @author Christoph Strobl
--- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensionsUnitTests.kt
+++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensionsUnitTests.kt
@@ -18,16 +18,20 @@ package org.springframework.data.redis.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
 import org.springframework.data.domain.Range
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate
 import org.springframework.data.redis.connection.RedisZSetCommands.Weights
+import org.springframework.data.redis.core.ZSetOperations.*
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono

 /**
- * Unit tests for [ReactiveZSetOperationsExtensions].
+ * Unit tests for `ReactiveZSetOperationsExtensions`.
 *
 * @author Mark Paluch
 * @author Christoph Strobl
@@ -154,6 +158,146 @@ class ReactiveZSetOperationsExtensionsUnitTests {
 		}
 	}

+	@Test
+	@ExperimentalCoroutinesApi
+	fun range() {
+
+		val range = Range.unbounded<Long>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.range(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.rangeAsFlow("foo", range).toList()).contains("bar")
+		}
+
+		verify {
+			operations.range("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun rangeWithScores() {
+
+		val tuple = mockk<TypedTuple<String>>(relaxed = true)
+		val range = Range.unbounded<Long>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.rangeWithScores(any(), any()) } returns Flux.just(tuple)
+
+		runBlocking {
+			assertThat(operations.rangeWithScoresAsFlow("foo", range).toList()).contains(tuple)
+		}
+
+		verify {
+			operations.rangeWithScores("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun rangeByScore() {
+
+		val range = Range.unbounded<Double>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.rangeByScore(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.rangeByScoreAsFlow("foo", range).toList()).contains("bar")
+		}
+
+		verify {
+			operations.rangeByScore("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun rangeByScoreWithScores() {
+
+		val tuple = mockk<TypedTuple<String>>(relaxed = true)
+		val range = Range.unbounded<Double>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.rangeByScoreWithScores(any(), any()) } returns Flux.just(tuple)
+
+		runBlocking {
+			assertThat(operations.rangeByScoreWithScoresAsFlow("foo", range).toList()).contains(tuple)
+		}
+
+		verify {
+			operations.rangeByScoreWithScores("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun reverseRange() {
+
+		val range = Range.unbounded<Long>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.reverseRange(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.reverseRangeAsFlow("foo", range).toList()).contains("bar")
+		}
+
+		verify {
+			operations.reverseRange("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun reverseRangeWithScores() {
+
+		val tuple = mockk<TypedTuple<String>>(relaxed = true)
+		val range = Range.unbounded<Long>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.reverseRangeWithScores(any(), any()) } returns Flux.just(tuple)
+
+		runBlocking {
+			assertThat(operations.reverseRangeWithScoresAsFlow("foo", range).toList()).contains(tuple)
+		}
+
+		verify {
+			operations.reverseRangeWithScores("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun reverseRangeByScore() {
+
+		val range = Range.unbounded<Double>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.reverseRangeByScore(any(), any()) } returns Flux.just("bar")
+
+		runBlocking {
+			assertThat(operations.reverseRangeByScoreAsFlow("foo", range).toList()).contains("bar")
+		}
+
+		verify {
+			operations.reverseRangeByScore("foo", range)
+		}
+	}
+
+	@Test
+	@ExperimentalCoroutinesApi
+	fun reverseRangeByScoreWithScores() {
+
+		val tuple = mockk<TypedTuple<String>>(relaxed = true)
+		val range = Range.unbounded<Double>()
+		val operations = mockk<ReactiveZSetOperations<String, String>>()
+		every { operations.reverseRangeByScoreWithScores(any(), any()) } returns Flux.just(tuple)
+
+		runBlocking {
+			assertThat(operations.reverseRangeByScoreWithScoresAsFlow("foo", range).toList()).contains(tuple)
+		}
+
+		verify {
+			operations.reverseRangeByScoreWithScores("foo", range)
+		}
+	}
+
 	@Test // DATAREDIS-937
 	fun count() {