From 78b895feb49bcb6d37aca56829b9b83caf7348ec Mon Sep 17 00:00:00 2001 From: Sebastien Deleuze Date: Tue, 10 Sep 2019 06:26:30 +0200 Subject: [PATCH] DATAREDIS-1033 - Add Kotlin Flow based extensions. Original Pull Request: #477 --- .../core/ReactiveGeoOperationsExtensions.kt | 54 +++- .../core/ReactiveHashOperationsExtensions.kt | 43 +++ .../core/ReactiveListOperationsExtensions.kt | 13 + .../core/ReactiveRedisOperationsExtensions.kt | 90 ++++++ .../core/ReactiveSetOperationsExtensions.kt | 142 ++++++++++ .../ReactiveStreamOperationsExtensions.kt | 138 ++++++++++ .../core/ReactiveZSetOperationsExtensions.kt | 87 +++++- ...eactiveGeoOperationsExtensionsUnitTests.kt | 128 ++++++++- ...activeHashOperationsExtensionsUnitTests.kt | 70 ++++- ...yperLogLogOperationsExtensionsUnitTests.kt | 2 +- ...activeListOperationsExtensionsUnitTests.kt | 21 +- ...ctiveRedisOperationsExtensionsUnitTests.kt | 149 +++++++++- ...eactiveSetOperationsExtensionsUnitTests.kt | 219 ++++++++++++++- ...tiveStreamOperationsExtensionsUnitTests.kt | 259 +++++++++++++++++- ...ctiveValueOperationsExtensionsUnitTests.kt | 2 +- ...activeZSetOperationsExtensionsUnitTests.kt | 146 +++++++++- 16 files changed, 1541 insertions(+), 22 deletions(-) diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensions.kt index 25dee9546..1df9279de 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensions.kt @@ -15,12 +15,19 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle +import kotlinx.coroutines.reactor.asFlux +import org.springframework.data.geo.Circle import org.springframework.data.geo.Distance +import org.springframework.data.geo.GeoResult import org.springframework.data.geo.Metric import org.springframework.data.geo.Point -import org.springframework.data.redis.connection.RedisGeoCommands +import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation +import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs /** * Coroutines variant of [ReactiveGeoOperations.add]. @@ -37,7 +44,7 @@ suspend fun ReactiveGeoOperations.addAndAwait(key: K, p * @author Mark Paluch * @since 2.2 */ -suspend fun ReactiveGeoOperations.addAndAwait(key: K, location: RedisGeoCommands.GeoLocation): Long = +suspend fun ReactiveGeoOperations.addAndAwait(key: K, location: GeoLocation): Long = add(key, location).awaitSingle() /** @@ -55,9 +62,19 @@ suspend fun ReactiveGeoOperations.addAndAwait(key: K, m * @author Mark Paluch * @since 2.2 */ -suspend fun ReactiveGeoOperations.addAndAwait(key: K, locations: Iterable>): Long = +suspend fun ReactiveGeoOperations.addAndAwait(key: K, locations: Iterable>): Long = add(key, locations).awaitSingle() +/** + * Coroutines [Flow] variant of [ReactiveGeoOperations.add]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveGeoOperations.add(key: K, locations: Flow>>): Flow = + add(key, locations.asFlux()).asFlow() + /** * Coroutines variant of [ReactiveGeoOperations.distance]. * @@ -115,6 +132,37 @@ suspend fun ReactiveGeoOperations.positionAndAwait(key: suspend fun ReactiveGeoOperations.positionAndAwait(key: K, vararg members: M): List = position(key, *members).awaitSingle() +/** + * Coroutines [Flow] variant of [ReactiveGeoOperations.radius]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveGeoOperations.radiusAsFlow(key: K, within: Circle, args: GeoRadiusCommandArgs? = null): Flow>> = + (if (args != null) radius(key, within, args) else radius(key, within)).asFlow() + + +/** + * Coroutines [Flow] variant of [ReactiveGeoOperations.radius]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveGeoOperations.radiusAsFlow(key: K, member: M, radius: Double): Flow>> = + radius(key, member, radius).asFlow() + +/** + * Coroutines [Flow] variant of [ReactiveGeoOperations.radius]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveGeoOperations.radiusAsFlow(key: K, member: M, distance: Distance, args: GeoRadiusCommandArgs? = null): Flow>> = + (if (args != null) radius(key, member, distance, args) else radius(key, member, distance)).asFlow() + /** * Coroutines variant of [ReactiveGeoOperations.remove]. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensions.kt index b1953bbae..648814fdd 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensions.kt @@ -15,6 +15,9 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle @@ -55,6 +58,16 @@ suspend fun ReactiveHashOperations.mult suspend fun ReactiveHashOperations.incrementAndAwait(key: H, hashKey: HK, delta: Long): Long = increment(key, hashKey, delta).awaitSingle() +/** + * Coroutines variant of [ReactiveHashOperations.keys]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveHashOperations.keysAsFlow(key: H): Flow = + keys(key).asFlow() + /** * Coroutines variant of [ReactiveHashOperations.increment]. * @@ -100,6 +113,36 @@ suspend fun ReactiveHashOperations.putA suspend fun ReactiveHashOperations.putIfAbsentAndAwait(key: H, hashKey: HK, hashValue: HV): Boolean = putIfAbsent(key, hashKey, hashValue).awaitSingle() +/** + * Coroutines variant of [ReactiveHashOperations.values]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveHashOperations.valuesAsFlow(key: H): Flow = + values(key).asFlow() + +/** + * Coroutines variant of [ReactiveHashOperations.entries]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveHashOperations.entriesAsFlow(key: H): Flow> = + entries(key).asFlow() + +/** + * Coroutines variant of [ReactiveHashOperations.scan]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveHashOperations.scanAsFlow(key: H, options: ScanOptions = ScanOptions.NONE): Flow> = + scan(key, options).asFlow() + /** * Coroutines variant of [ReactiveHashOperations.remove]. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensions.kt index 6b770791a..93d3ccf3c 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensions.kt @@ -15,10 +15,23 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle import java.time.Duration +/** + * Coroutines variant of [ReactiveListOperations.range]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveListOperations.rangeAsFlow(key: K, start: Long, end: Long): Flow = + range(key, start, end).asFlow() + /** * Coroutines variant of [ReactiveListOperations.trim]. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt index 5618e2f4f..ddbefd330 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt @@ -15,12 +15,52 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow +import kotlinx.coroutines.reactive.asPublisher import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle import org.springframework.data.redis.connection.DataType +import org.springframework.data.redis.connection.ReactiveRedisConnection +import org.springframework.data.redis.connection.ReactiveSubscription.* +import org.springframework.data.redis.core.script.RedisScript +import org.springframework.data.redis.listener.Topic +import org.springframework.data.redis.serializer.RedisElementReader +import org.springframework.data.redis.serializer.RedisElementWriter import java.time.Duration import java.time.Instant +/** + * Coroutines variant of [ReactiveRedisOperations.execute]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.executeAsFlow(action: (ReactiveRedisConnection) -> Flow): Flow = + execute { action(it).asPublisher() }.asFlow() + +/** + * Coroutines variant of [ReactiveRedisOperations.execute]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.executeAsFlow(script: RedisScript, keys: List = emptyList(), args: List<*> = emptyList()): Flow = + execute(script, keys, args).asFlow() + +/** + * Coroutines variant of [ReactiveRedisOperations.execute]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.executeAsFlow(script: RedisScript, keys: List = emptyList(), args: List<*> = emptyList(), argsWriter: RedisElementWriter<*>, resultReader: RedisElementReader): Flow = + execute(script, keys, args, argsWriter, resultReader).asFlow() + /** * Coroutines variant of [ReactiveRedisOperations.convertAndSend]. * @@ -30,6 +70,36 @@ import java.time.Instant suspend fun ReactiveRedisOperations.sendAndAwait(destination: String, message: V): Long = convertAndSend(destination, message).awaitSingle() +/** + * Coroutines variant of [ReactiveRedisOperations.listenToChannel]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.listenToChannelAsFlow(vararg channels: String): Flow> = + listenToChannel(*channels).asFlow() + +/** + * Coroutines variant of [ReactiveRedisOperations.listenToPattern]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.listenToPatternAsFlow(vararg patterns: String): Flow> = + listenToPattern(*patterns).asFlow() + +/** + * Coroutines variant of [ReactiveRedisOperations.listenTo]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.listenToAsFlow(vararg topics: Topic): Flow> = + listenTo(*topics).asFlow() + /** * Coroutines variant of [ReactiveRedisOperations.hasKey]. * @@ -48,6 +118,26 @@ suspend fun ReactiveRedisOperations.hasKeyAndAwait(key: suspend fun ReactiveRedisOperations.typeAndAwait(key: K): DataType = type(key).awaitSingle() +/** + * Coroutines variant of [ReactiveRedisOperations.keys]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.keysAsFlow(pattern: K): Flow = + keys(pattern).asFlow() + +/** + * Coroutines variant of [ReactiveRedisOperations.scan]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveRedisOperations.scanAsFlow(options: ScanOptions = ScanOptions.NONE): Flow = + scan(options).asFlow() + /** * Coroutines variant of [ReactiveRedisOperations.randomKey]. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensions.kt index 5dcdd9dcd..8826424ef 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensions.kt @@ -15,6 +15,9 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle @@ -45,6 +48,16 @@ suspend fun ReactiveSetOperations.removeAndAwait(key: K suspend fun ReactiveSetOperations.popAndAwait(key: K): V? = pop(key).awaitFirstOrNull() +/** + * Coroutines variant of [ReactiveSetOperations.pop]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.popAsFlow(key: K, count: Long): Flow = + pop(key, count).asFlow() + /** * Coroutines variant of [ReactiveSetOperations.move]. * @@ -72,6 +85,36 @@ suspend fun ReactiveSetOperations.sizeAndAwait(key: K): suspend fun ReactiveSetOperations.isMemberAndAwait(key: K, value: V): Boolean = isMember(key, value).awaitSingle() +/** + * Coroutines variant of [ReactiveSetOperations.intersect]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.intersectAsFlow(key: K, otherKey: K): Flow = + intersect(key, otherKey).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.intersect]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.intersectAsFlow(key: K, otherKeys: Collection): Flow = + intersect(key, otherKeys).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.intersect]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.intersectAsFlow(otherKeys: Collection): Flow = + intersect(otherKeys).asFlow() + /** * Coroutines variant of [ReactiveSetOperations.intersectAndStore]. * @@ -90,6 +133,36 @@ suspend fun ReactiveSetOperations.intersectAndStoreAndA suspend fun ReactiveSetOperations.intersectAndStoreAndAwait(keys: Collection, destKey: K): Long = intersectAndStore(keys, destKey).awaitSingle() +/** + * Coroutines variant of [ReactiveSetOperations.union]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.unionAsFlow(key: K, otherKey: K): Flow = + union(key, otherKey).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.union]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.unionAsFlow(key: K, otherKeys: Collection): Flow = + union(key, otherKeys).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.union]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.unionAsFlow(otherKeys: Collection): Flow = + union(otherKeys).asFlow() + /** * Coroutines variant of [ReactiveSetOperations.unionAndStore]. * @@ -108,6 +181,36 @@ suspend fun ReactiveSetOperations.unionAndStoreAndAwait suspend fun ReactiveSetOperations.unionAndStoreAndAwait(keys: Collection, destKey: K): Long = unionAndStore(keys, destKey).awaitSingle() +/** + * Coroutines variant of [ReactiveSetOperations.difference]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.differenceAsFlow(key: K, otherKey: K): Flow = + difference(key, otherKey).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.difference]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.differenceAsFlow(key: K, otherKeys: Collection): Flow = + difference(key, otherKeys).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.difference]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.differenceAsFlow(otherKeys: Collection): Flow = + difference(otherKeys).asFlow() + /** * Coroutines variant of [ReactiveSetOperations.differenceAndStore]. * @@ -126,6 +229,25 @@ suspend fun ReactiveSetOperations.differenceAndStoreAnd suspend fun ReactiveSetOperations.differenceAndStoreAndAwait(keys: Collection, destKey: K): Long = differenceAndStore(keys, destKey).awaitSingle() +/** + * Coroutines variant of [ReactiveSetOperations.members]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.membersAsFlow(key: K): Flow = + members(key).asFlow() + +/** + * Coroutines variant of [ReactiveHashOperations.scan]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.scanAsFlow(key: K, options: ScanOptions = ScanOptions.NONE): Flow = + scan(key, options).asFlow() /** * Coroutines variant of [ReactiveSetOperations.randomMember]. * @@ -135,6 +257,26 @@ suspend fun ReactiveSetOperations.differenceAndStoreAnd suspend fun ReactiveSetOperations.randomMemberAndAwait(key: K): V? = randomMember(key).awaitFirstOrNull() +/** + * Coroutines variant of [ReactiveSetOperations.distinctRandomMembers]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.distinctRandomMembersAsFlow(key: K, count: Long): Flow = + distinctRandomMembers(key, count).asFlow() + +/** + * Coroutines variant of [ReactiveSetOperations.randomMembers]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveSetOperations.randomMembersAsFlow(key: K, count: Long): Flow = + randomMembers(key, count).asFlow() + /** * Coroutines variant of [ReactiveSetOperations.delete]. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt index 6e7dbf89a..0ef6abdcc 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt @@ -15,7 +15,13 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow +import kotlinx.coroutines.reactive.asPublisher import kotlinx.coroutines.reactive.awaitSingle +import org.springframework.data.domain.Range +import org.springframework.data.redis.connection.RedisZSetCommands.* import org.springframework.data.redis.connection.stream.* /** @@ -45,6 +51,16 @@ suspend fun ReactiveStreamOperations.ac suspend fun ReactiveStreamOperations.acknowledgeAndAwait(group: String, record: Record): Long = acknowledge(group, record).awaitSingle() +/** + * Coroutines variant of [ReactiveStreamOperations.add]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.add(key: K, bodyFlow: Flow>): Flow = + add(key, bodyFlow.asPublisher()).asFlow() + /** * Coroutines variant of [ReactiveStreamOperations.add]. * @@ -135,6 +151,128 @@ suspend fun ReactiveStreamOperations.de suspend fun ReactiveStreamOperations.sizeAndAwait(key: K): Long = size(key).awaitSingle() + +/** + * Coroutines variant of [ReactiveStreamOperations.range]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.rangeAsFlow(key: K, range: Range, limit: Limit = Limit.unlimited()): Flow> + = range(key, range, limit).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.range]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +inline fun ReactiveStreamOperations.rangeWithTypeAsFlow(key: K, range: Range, limit: Limit = Limit.unlimited()): Flow> + = range(V::class.java, key, range, limit).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.readAsFlow(vararg stream: StreamOffset): Flow> = + read(*stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.readAsFlow(readOptions: StreamReadOptions, vararg stream: StreamOffset): Flow> = + read(readOptions, *stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +inline fun ReactiveStreamOperations.readWithTypeAsFlow(vararg stream: StreamOffset): Flow> = + read(V::class.java, *stream).asFlow() + + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +inline fun ReactiveStreamOperations.readWithTypeAsFlow(readOptions: StreamReadOptions, vararg stream: StreamOffset): Flow> = + read(V::class.java, readOptions, *stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.readAsFlow(consumer: Consumer, vararg stream: StreamOffset): Flow> = + read(consumer, *stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.readAsFlow(consumer: Consumer, readOptions: StreamReadOptions, vararg stream: StreamOffset): Flow> = + read(consumer, readOptions, *stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +inline fun ReactiveStreamOperations.readWithTypeAsFlow(consumer: Consumer, vararg stream: StreamOffset): Flow> = + read(V::class.java, consumer, *stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.read]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +inline fun ReactiveStreamOperations.readWithTypeAsFlow(consumer: Consumer, readOptions: StreamReadOptions, vararg stream: StreamOffset): Flow> = + read(V::class.java, consumer, readOptions, *stream).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.reverseRange]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveStreamOperations.reverseRangeAsFlow(key: K, range: Range, limit: Limit = Limit.unlimited()): Flow> + = reverseRange(key, range, limit).asFlow() + +/** + * Coroutines variant of [ReactiveStreamOperations.reverseRange]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +inline fun ReactiveStreamOperations.reverseRangeWithTypeAsFlow(key: K, range: Range, limit: Limit = Limit.unlimited()): Flow> = + reverseRange(V::class.java, key, range, limit).asFlow() + /** * Coroutines variant of [ReactiveStreamOperations.trim]. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensions.kt index 942bde7e9..d77ed679b 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensions.kt @@ -15,10 +15,15 @@ */ package org.springframework.data.redis.core +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle import org.springframework.data.domain.Range import org.springframework.data.redis.connection.RedisZSetCommands +import org.springframework.data.redis.connection.RedisZSetCommands.Limit +import org.springframework.data.redis.core.ZSetOperations.* /** * Coroutines variant of [ReactiveZSetOperations.add]. @@ -35,7 +40,7 @@ suspend fun ReactiveZSetOperations.addAndAwait(key: K, * @author Mark Paluch * @since 2.2 */ -suspend fun ReactiveZSetOperations.addAllAndAwait(key: K, values: Collection>): Long = +suspend fun ReactiveZSetOperations.addAllAndAwait(key: K, values: Collection>): Long = addAll(key, values).awaitSingle() /** @@ -74,6 +79,86 @@ suspend fun ReactiveZSetOperations.rankAndAwait(key: K, suspend fun ReactiveZSetOperations.reverseRankAndAwait(key: K, value: V): Long? = reverseRank(key, value).awaitFirstOrNull() +/** + * Coroutines variant of [ReactiveZSetOperations.range]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.rangeAsFlow(key: K, range: Range): Flow = + range(key, range).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.rangeWithScores]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.rangeWithScoresAsFlow(key: K, range: Range): Flow> = + rangeWithScores(key, range).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.rangeByScore]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.rangeByScoreAsFlow(key: K, range: Range, limit: Limit? = null): Flow = + (if (limit == null) rangeByScore(key, range) else rangeByScore(key, range, limit)).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.rangeByScoreWithScores]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.rangeByScoreWithScoresAsFlow(key: K, range: Range, limit: Limit? = null): Flow> = + (if (limit == null) rangeByScoreWithScores(key, range) else rangeByScoreWithScores(key, range, limit)).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.reverseRange]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.reverseRangeAsFlow(key: K, range: Range): Flow = + reverseRange(key, range).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.reverseRangeWithScores]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.reverseRangeWithScoresAsFlow(key: K, range: Range): Flow> = + reverseRangeWithScores(key, range).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.reverseRangeByScore]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.reverseRangeByScoreAsFlow(key: K, range: Range, limit: Limit? = null): Flow = + (if (limit == null) reverseRangeByScore(key, range) else reverseRangeByScore(key, range, limit)).asFlow() + +/** + * Coroutines variant of [ReactiveZSetOperations.reverseRangeByScoreWithScores]. + * + * @author Sebastien Deleuze + * @since 2.2 + */ +@ExperimentalCoroutinesApi +fun ReactiveZSetOperations.reverseRangeByScoreWithScoresAsFlow(key: K, range: Range, limit: Limit? = null): Flow> = + (if (limit == null) reverseRangeByScoreWithScores(key, range) else reverseRangeByScoreWithScores(key, range, limit)).asFlow() + /** * Coroutines variant of [ReactiveZSetOperations.count]. * diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveGeoOperationsExtensionsUnitTests.kt index d56f8683a..3f8f1a462 100644 --- 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>() - every { operations.add(any(), any>()) } returns Mono.just(1) + every { operations.add(any(), any>()) } 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>() - every { operations.add(any(), any>>()) } returns Mono.just(1) + every { operations.add(any(), any>>()) } 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>() + every { operations.add(any(), any>>>()) } 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>>>()) } } @@ -246,6 +271,95 @@ class ReactiveGeoOperationsExtensionsUnitTests { } } + @Test + @ExperimentalCoroutinesApi + fun radiusAsFlowCircle() { + val operations = mockk>() + 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>() + 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>() + val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0)) + + every { operations.radius(any(), any(), any()) } 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>() + val result = GeoResult(GeoLocation("bar", Point(1.0, 2.0)), Distance(1.0)) + val distance = Distance(2.0) + + every { operations.radius(any(), any(), any()) } 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>() + 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() { diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveHashOperationsExtensionsUnitTests.kt index d3d61bdba..31713baae 100644 --- 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>() + 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>() + 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>() + 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>() + 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() { diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveHyperLogLogOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveHyperLogLogOperationsExtensionsUnitTests.kt index 57ecf1f63..0019f1214 100644 --- 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 */ diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveListOperationsExtensionsUnitTests.kt index dd460a7b5..4ecd756d4 100644 --- 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>() + 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() { diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt index 85e861043..ab34974cf 100644 --- 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>() + every { operations.execute(any>()) } returns Flux.just("foo") + + runBlocking { + assertThat(operations.executeAsFlow { flow { emit("foo")} }.toList()).contains("foo") + } + + verify { + operations.execute(any>()) + } + } + + @Test + @ExperimentalCoroutinesApi + fun `execute with script`() { + + val script = RedisScript.of("foo") + val operations = mockk>() + every { operations.execute(any>(), 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("foo") + val argsWriter = mockk>(relaxed = true) + val resultReader = mockk>(relaxed = true) + val operations = mockk>() + every { operations.execute(any>(), 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>() + 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>() + 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>() + 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>() + 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>() + 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() { diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveSetOperationsExtensionsUnitTests.kt index 76aad2c66..4af0f9cf5 100644 --- 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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>() + 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() { diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt index c04228f2b..399e84105 100644 --- 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>() 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>() + val recordId = RecordId.of("0-0") + every { operations.add(any(), any>>()) } returns Flux.just(recordId) + + runBlocking { + val bodyFlow = flow { emit(map) } + assertThat(operations.add("foo", bodyFlow).toList()).contains(recordId) + } + + verify { + operations.add("foo", any>>()) + } + } + @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>() + 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>() + every { operations.range(any>(), any(), any(), any()) } returns Flux.just(record) + + runBlocking { + assertThat(operations.rangeWithTypeAsFlow("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>() + 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>() + 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>() + every { operations.read(String::class.java, offset1, offset2) } returns Flux.just(record) + + runBlocking { + assertThat(operations.readWithTypeAsFlow(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>() + every { operations.read(String::class.java, options, offset1, offset2) } returns Flux.just(record) + + runBlocking { + assertThat(operations.readWithTypeAsFlow(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>() + 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>() + 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>() + every { operations.read(String::class.java, consumer, offset1, offset2) } returns Flux.just(record) + + runBlocking { + assertThat(operations.readWithTypeAsFlow(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>() + every { operations.read(String::class.java, consumer, options, offset1, offset2) } returns Flux.just(record) + + runBlocking { + assertThat(operations.readWithTypeAsFlow(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>() + 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>() + every { operations.reverseRange(any>(), any(), any(), any()) } returns Flux.just(record) + + runBlocking { + assertThat(operations.reverseRangeWithTypeAsFlow("foo", range).toList()).contains(record) + } + + verify { + operations.reverseRange(String::class.java, "foo", range, Limit.unlimited()) + } + } + @Test // DATAREDIS-937 fun trim() { diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveValueOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveValueOperationsExtensionsUnitTests.kt index d12e18dc4..c91ff51b5 100644 --- 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 diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveZSetOperationsExtensionsUnitTests.kt index 94af28ad8..e353bede2 100644 --- 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() + val operations = mockk>() + 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>(relaxed = true) + val range = Range.unbounded() + val operations = mockk>() + 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() + val operations = mockk>() + 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>(relaxed = true) + val range = Range.unbounded() + val operations = mockk>() + 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() + val operations = mockk>() + 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>(relaxed = true) + val range = Range.unbounded() + val operations = mockk>() + 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() + val operations = mockk>() + 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>(relaxed = true) + val range = Range.unbounded() + val operations = mockk>() + 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() {