diff --git a/src/main/asciidoc/new-features.adoc b/src/main/asciidoc/new-features.adoc index 6d18542fa..9c1ec669e 100644 --- a/src/main/asciidoc/new-features.adoc +++ b/src/main/asciidoc/new-features.adoc @@ -14,6 +14,7 @@ This section briefly covers items that are new and noteworthy in the latest rele * <> to send and receive a message stream. * `BITFIELD`, `BITPOS`, and `OBJECT` command support. * Align return types of `BoundZSetOperations` with `ZSetOperations`. +* Reactive `SCAN`, `HSCAN`, `SSCAN`, and `ZSCAN` support. [[new-in-2.0.0]] == New in Spring Data Redis 2.0 diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java index 4041d7ab0..267585d33 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java @@ -31,8 +31,10 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Boolean import org.springframework.data.redis.connection.ReactiveRedisConnection.Command; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; +import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -578,6 +580,45 @@ public interface ReactiveHashCommands { */ Flux>>> hGetAll(Publisher commands); + /** + * Use a {@link Flux} to iterate over entries in the hash at {@code key}. The resulting {@link Flux} acts as a cursor + * and issues {@code HSCAN} commands itself as long as the subscriber signals demand . + * + * @param key must not be {@literal null}. + * @return + * @see Redis Documentation: HSCAN + * @since 2.1 + */ + default Flux> hScan(ByteBuffer key) { + return hScan(key, ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over entries in the hash at {@code key} given {@link ScanOptions}. The resulting + * {@link Flux} acts as a cursor and issues {@code HSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @param options must not be {@literal null}. + * @return + * @see Redis Documentation: HSCAN + * @since 2.1 + */ + default Flux> hScan(ByteBuffer key, ScanOptions options) { + return hScan(Mono.just(KeyScanCommand.key(key).withOptions(options))).map(CommandResponse::getOutput) + .flatMap(it -> it); + } + + /** + * Use a {@link Flux} to iterate over entries in the hash at {@code key}. The resulting {@link Flux} acts as a cursor + * and issues {@code HSCAN} commands itself as long as the subscriber signals demand. + * + * @param commands must not be {@literal null}. + * @return + * @see Redis Documentation: HSCAN + * @since 2.1 + */ + Flux>>> hScan(Publisher commands); + /** * @author Christoph Strobl * @see Redis Documentation: HSTRLEN diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveKeyCommands.java index 182374eb9..a69fa0f8e 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveKeyCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveKeyCommands.java @@ -30,6 +30,7 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Command import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -133,6 +134,29 @@ public interface ReactiveKeyCommands { */ Flux> keys(Publisher patterns); + /** + * Use a {@link Flux} to iterate over keys. The resulting {@link Flux} acts as a cursor and issues {@code SCAN} + * commands itself as long as the subscriber signals demand. + * + * @return never {@literal null}. + * @see Redis Documentation: SCAN + * @since 2.1 + */ + default Flux scan() { + return scan(ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over keys. The resulting {@link Flux} acts as a cursor and issues {@code SCAN} + * commands itself as long as the subscriber signals demand. + * + * @param options must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: SCAN + * @since 2.1 + */ + Flux scan(ScanOptions options); + /** * Return a random key from the keyspace. * diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java index 415c99852..eeb275950 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java @@ -24,6 +24,7 @@ import java.util.List; import org.springframework.data.domain.Range; import org.springframework.data.domain.Range.Bound; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -209,6 +210,49 @@ public interface ReactiveRedisConnection extends Closeable { } } + /** + * {@link Command} for key-bound scan operations like {@code HSCAN}, {@code SSCAN}, and {@code ZSCAN}, . + * + * @author Mark Paluch + * @since 2.1 + */ + class KeyScanCommand extends KeyCommand { + + private ScanOptions options; + + private KeyScanCommand(@Nullable ByteBuffer key, ScanOptions options) { + + super(key); + + Assert.notNull(options, "ScanOptions must not be null!"); + this.options = options; + } + + /** + * Creates a new {@link KeyScanCommand} given a {@code key}. + * + * @param key must not be {@literal null}. + * @return a new {@link KeyScanCommand} for {@code key}. + */ + public static KeyScanCommand key(ByteBuffer key) { + return new KeyScanCommand(key, ScanOptions.NONE); + } + + /** + * Applies {@link ScanOptions}. Constructs a new command instance with all previously configured properties. + * + * @param options must not be {@literal null}. + * @return a new {@link KeyScanCommand} with {@link ScanOptions} applied. + */ + public KeyScanCommand withOptions(ScanOptions options) { + return new KeyScanCommand(getKey(), options); + } + + public ScanOptions getOptions() { + return options; + } + } + /** * @author Christoph Strobl */ diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveSetCommands.java index fe323a64a..bfc982c30 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveSetCommands.java @@ -31,7 +31,9 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBuf import org.springframework.data.redis.connection.ReactiveRedisConnection.Command; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; +import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -1013,6 +1015,45 @@ public interface ReactiveSetCommands { */ Flux>> sMembers(Publisher commands); + /** + * Use a {@link Flux} to iterate over members in the set at {@code key}. The resulting {@link Flux} acts as a cursor + * and issues {@code SSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @return + * @see Redis Documentation: SSCAN + * @since 2.1 + */ + default Flux sScan(ByteBuffer key) { + return sScan(key, ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over members in the set at {@code key} given {@link ScanOptions}. The resulting + * {@link Flux} acts as a cursor and issues {@code SSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @param options must not be {@literal null}. + * @return + * @see Redis Documentation: SSCAN + * @since 2.1 + */ + default Flux sScan(ByteBuffer key, ScanOptions options) { + return sScan(Mono.just(KeyScanCommand.key(key).withOptions(options))).map(CommandResponse::getOutput) + .flatMap(it -> it); + } + + /** + * Use a {@link Flux} to iterate over members in the set at {@code key}. The resulting {@link Flux} acts as a cursor + * and issues {@code SSCAN} commands itself as long as the subscriber signals demand. + * + * @param commands must not be {@literal null}. + * @return + * @see Redis Documentation: SSCAN + * @since 2.1 + */ + Flux>> sScan(Publisher commands); + /** * {@code SRANDMEMBER} command parameters. * diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java index b3e2b60ff..8f2a5768b 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java @@ -30,11 +30,13 @@ import org.springframework.data.domain.Range; import org.springframework.data.domain.Sort.Direction; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; +import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; import org.springframework.data.redis.connection.RedisZSetCommands.Weights; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -997,6 +999,45 @@ public interface ReactiveZSetCommands { */ Flux>> zRangeByScore(Publisher commands); + /** + * Use a {@link Flux} to iterate over members in the sorted set at {@code key}. The resulting {@link Flux} acts as a + * cursor and issues {@code ZSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @return + * @see Redis Documentation: ZSCAN + * @since 2.1 + */ + default Flux zScan(ByteBuffer key) { + return zScan(key, ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over members in the sorted set at {@code key} given {@link ScanOptions}. The + * resulting {@link Flux} acts as a cursor and issues {@code ZSCAN} commands itself as long as the subscriber signals. + * + * @param key must not be {@literal null}. + * @param options must not be {@literal null}. + * @return + * @see Redis Documentation: ZSCAN + * @since 2.1 + */ + default Flux zScan(ByteBuffer key, ScanOptions options) { + return zScan(Mono.just(KeyScanCommand.key(key).withOptions(options))).map(CommandResponse::getOutput) + .flatMap(it -> it); + } + + /** + * Use a {@link Flux} to iterate over members in the sorted set at {@code key}. The resulting {@link Flux} acts as a + * cursor and issues {@code ZSCAN} commands itself as long as the subscriber signals demand. + * + * @param commands must not be {@literal null}. + * @return + * @see Redis Documentation: ZSCAN + * @since 2.1 + */ + Flux>> zScan(Publisher commands); + /** * {@code ZCOUNT} command parameters. * diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java index 4137eea35..aa6d3d4d5 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java @@ -204,7 +204,7 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands { @Override protected LettuceScanIteration doScan(io.lettuce.core.ScanCursor cursor, ScanOptions options) { - ScanArgs scanArgs = connection.getScanArgs(options); + ScanArgs scanArgs = LettuceConverters.toScanArgs(options); KeyScanCursor keyScanCursor = client.scan(cursor, scanArgs); return new LettuceScanIteration<>(keyScanCursor, keyScanCursor.getKeys()); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index f3b0c4e47..2e2f08375 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -23,7 +23,6 @@ import io.lettuce.core.RedisClient; import io.lettuce.core.RedisException; import io.lettuce.core.RedisFuture; import io.lettuce.core.RedisURI; -import io.lettuce.core.ScanArgs; import io.lettuce.core.TransactionResult; import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.StatefulRedisConnection; @@ -68,7 +67,6 @@ import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvid import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceResultBuilder; import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceStatusResult; import org.springframework.data.redis.core.RedisCommand; -import org.springframework.data.redis.core.ScanOptions; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -922,26 +920,6 @@ public class LettuceConnection extends AbstractRedisConnection { return io.lettuce.core.ScanCursor.of(Long.toString(cursorId)); } - @Nullable - ScanArgs getScanArgs(@Nullable ScanOptions options) { - - if (options == null) { - return null; - } - - ScanArgs scanArgs = new ScanArgs(); - - if (options.getPattern() != null) { - scanArgs.match(options.getPattern()); - } - - if (options.getCount() != null) { - scanArgs.limit(options.getCount()); - } - - return scanArgs; - } - private void validateCommandIfRunningInTransactionMode(CommandType cmd, byte[]... args) { if (this.isQueueing()) { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java index 1bd9a2529..630603de1 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java @@ -63,6 +63,7 @@ import org.springframework.data.redis.connection.convert.Converters; import org.springframework.data.redis.connection.convert.ListConverter; import org.springframework.data.redis.connection.convert.LongToBooleanConverter; import org.springframework.data.redis.connection.convert.StringToRedisClientInfoConverter; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.data.redis.core.types.Expiration; import org.springframework.data.redis.core.types.RedisClientInfo; import org.springframework.data.redis.util.ByteUtils; @@ -878,6 +879,33 @@ abstract public class LettuceConverters extends Converters { return args; } + /** + * Convert {@link ScanOptions} to {@link ScanArgs}. + * + * @param options the {@link ScanOptions} to convert, may be {@literal null}. + * @return the converted {@link ScanArgs}. Returns {@literal null} if {@link ScanOptions} is {@literal null}. + * @see 2.1 + */ + @Nullable + static ScanArgs toScanArgs(@Nullable ScanOptions options) { + + if (options == null) { + return null; + } + + ScanArgs scanArgs = new ScanArgs(); + + if (options.getPattern() != null) { + scanArgs.match(options.getPattern()); + } + + if (options.getCount() != null) { + scanArgs.limit(options.getCount()); + } + + return scanArgs; + } + /** * Get {@link Converter} capable of {@link Set} of {@link Byte} into {@link GeoResults}. * diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceHashCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceHashCommands.java index efb4ecb00..1f0e5553b 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceHashCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceHashCommands.java @@ -401,7 +401,7 @@ class LettuceHashCommands implements RedisHashCommands { } io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId); - ScanArgs scanArgs = connection.getScanArgs(options); + ScanArgs scanArgs = LettuceConverters.toScanArgs(options); MapScanCursor mapScanCursor = getConnection().hscan(key, scanCursor, scanArgs); String nextCursorId = mapScanCursor.getCursor(); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceKeyCommands.java index 837b97f97..daa1f1e3e 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceKeyCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceKeyCommands.java @@ -259,7 +259,7 @@ class LettuceKeyCommands implements RedisKeyCommands { throw new UnsupportedOperationException("'SCAN' cannot be called in pipeline / transaction mode."); } - ScanArgs scanArgs = connection.getScanArgs(options); + ScanArgs scanArgs = LettuceConverters.toScanArgs(options); KeyScanCursor keyScanCursor = getConnection().scan(cursor, scanArgs); List keys = keyScanCursor.getKeys(); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommands.java index 39bed5d57..dd98592aa 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommands.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.KeyValue; +import io.lettuce.core.ScanStream; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -31,6 +32,7 @@ import org.springframework.data.redis.connection.ReactiveHashCommands; import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; +import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.util.Assert; @@ -214,6 +216,44 @@ class LettuceReactiveHashCommands implements ReactiveHashCommands { })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveHashCommands#hScan(org.reactivestreams.Publisher) + */ + @Override + public Flux>>> hScan( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Key must not be null!"); + Assert.notNull(command.getOptions(), "ScanOptions must not be null!"); + + Flux> result = ScanStream.hscan(cmd, command.getKey(), + LettuceConverters.toScanArgs(command.getOptions())); + + Flux> entryFlux = result.map(it -> new Entry() { + + @Override + public ByteBuffer getKey() { + return it.getKey(); + } + + @Override + public ByteBuffer getValue() { + return it.getValue(); + } + + @Override + public ByteBuffer setValue(ByteBuffer value) { + throw new UnsupportedOperationException("Cannot set value for entry in cursor"); + } + }); + + return Mono.just(new CommandResponse<>(command, entryFlux)); + })); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveHashCommands#hstrlen(org.reactivestreams.Publisher) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommands.java index 7a454258a..e336e9197 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommands.java @@ -15,6 +15,7 @@ */ package org.springframework.data.redis.connection.lettuce; +import io.lettuce.core.ScanStream; import io.lettuce.core.api.reactive.RedisKeyReactiveCommands; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -34,6 +35,7 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiVa import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.data.redis.connection.ValueEncoding; import org.springframework.data.redis.connection.ValueEncoding.RedisValueEncoding; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.util.Assert; /** @@ -119,6 +121,18 @@ class LettuceReactiveKeyCommands implements ReactiveKeyCommands { })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#scan(org.springframework.data.redis.core.ScanOptions) + */ + @Override + public Flux scan(ScanOptions options) { + + Assert.notNull(options, "ScanOptions must not be null!"); + + return connection.execute(cmd -> ScanStream.scan(cmd, LettuceConverters.toScanArgs(options))); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#randomKey() diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommands.java index 816712dfe..c3a2eaac2 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommands.java @@ -15,6 +15,7 @@ */ package org.springframework.data.redis.connection.lettuce; +import io.lettuce.core.ScanStream; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -25,6 +26,7 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Boolean import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; +import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.data.redis.connection.ReactiveSetCommands; import org.springframework.util.Assert; @@ -276,6 +278,25 @@ class LettuceReactiveSetCommands implements ReactiveSetCommands { })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveSetCommands#sScan(org.reactivestreams.Publisher) + */ + @Override + public Flux>> sScan(Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Key must not be null!"); + Assert.notNull(command.getOptions(), "ScanOptions must not be null!"); + + Flux result = ScanStream.sscan(cmd, command.getKey(), + LettuceConverters.toScanArgs(command.getOptions())); + + return Mono.just(new CommandResponse<>(command, result)); + })); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveSetCommands#sRandMembers(org.reactivestreams.Publisher) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java index 41a7f5ae4..3f9328d02 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java @@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.Range; import io.lettuce.core.Range.Boundary; +import io.lettuce.core.ScanStream; import io.lettuce.core.ScoredValue; import io.lettuce.core.ZAddArgs; import io.lettuce.core.ZStoreArgs; @@ -34,6 +35,7 @@ import org.springframework.data.domain.Sort.Direction; import org.springframework.data.redis.connection.DefaultTuple; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; +import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.data.redis.connection.ReactiveZSetCommands; import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; @@ -292,6 +294,25 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zScan(org.reactivestreams.Publisher) + */ + @Override + public Flux>> zScan(Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Key must not be null!"); + Assert.notNull(command.getOptions(), "ScanOptions must not be null!"); + + Flux result = ScanStream.zscan(cmd, command.getKey(), LettuceConverters.toScanArgs(command.getOptions())) + .map(it -> new DefaultTuple(ByteUtils.getBytes(it.getValue()), it.getScore())); + + return Mono.just(new CommandResponse<>(command, result)); + })); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zCount(org.reactivestreams.Publisher) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSetCommands.java index 1a54ccf34..8fb37ed46 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSetCommands.java @@ -476,7 +476,7 @@ class LettuceSetCommands implements RedisSetCommands { } io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId); - ScanArgs scanArgs = connection.getScanArgs(options); + ScanArgs scanArgs = LettuceConverters.toScanArgs(options); ValueScanCursor valueScanCursor = getConnection().sscan(key, scanCursor, scanArgs); String nextCursorId = valueScanCursor.getCursor(); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java index e036e622c..35a51f95d 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java @@ -711,7 +711,7 @@ class LettuceZSetCommands implements RedisZSetCommands { } io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId); - ScanArgs scanArgs = connection.getScanArgs(options); + ScanArgs scanArgs = LettuceConverters.toScanArgs(options); ScoredValueScanCursor scoredValueScanCursor = getConnection().zscan(key, scanCursor, scanArgs); String nextCursorId = scoredValueScanCursor.getCursor(); diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java index eab64c7a4..1d1e820fa 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java @@ -221,6 +221,19 @@ class DefaultReactiveHashOperations implements ReactiveHashOperations .map(this::deserializeHashEntry)); } + /* (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveHashOperations#scan(java.lang.Object, org.springframework.data.redis.core.ScanOptions) + */ + @Override + public Flux> scan(H key, ScanOptions options) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(key, "ScanOptions must not be null!"); + + return createFlux(connection -> connection.hScan(rawKey(key), options) // + .map(this::deserializeHashEntry)); + } + /* (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveHashOperations#delete(java.lang.Object) */ diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java index 348026c5a..f92f97e76 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java @@ -322,6 +322,18 @@ class DefaultReactiveSetOperations implements ReactiveSetOperations return createFlux(connection -> connection.sMembers(rawKey(key)).map(this::readValue)); } + /* (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveSetOperations#scan(java.lang.Object, org.springframework.data.redis.core.ScanOptions) + */ + @Override + public Flux scan(K key, ScanOptions options) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(options, "ScanOptions must not be null!"); + + return createFlux(connection -> connection.sScan(rawKey(key)).map(this::readValue)); + } + /* (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveSetOperations#randomMember(java.lang.Object) */ diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java index 6c9c46f84..e7ae40882 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java @@ -283,6 +283,18 @@ class DefaultReactiveZSetOperations implements ReactiveZSetOperations connection.zRevRangeByScoreWithScores(rawKey(key), range, limit).map(this::readTypedTuple)); } + /* (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#scan(java.lang.Object, org.springframework.data.redis.core.ScanOptions) + */ + @Override + public Flux> scan(K key, ScanOptions options) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(options, "ScanOptions must not be null!"); + + return createFlux(connection -> connection.zScan(rawKey(key), options).map(this::readTypedTuple)); + } + /* (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveZSetOperations#count(java.lang.Object, org.springframework.data.domain.Range) */ diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveHashOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveHashOperations.java index d06a00990..b3cd3df50 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveHashOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveHashOperations.java @@ -146,6 +146,31 @@ public interface ReactiveHashOperations { */ Flux> entries(H key); + /** + * Use a {@link Flux} to iterate over entries in the hash at {@code key} given {@link ScanOptions}. The resulting + * {@link Flux} acts as a cursor and issues {@code HSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: HSCAN + * @since 2.1 + */ + default Flux> scan(H key) { + return scan(key, ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over entries in the hash at {@code key} given {@link ScanOptions}. The resulting + * {@link Flux} acts as a cursor and issues {@code HSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @param options must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: HSCAN + * @since 2.1 + */ + Flux> scan(H key, ScanOptions options); + /** * Removes the given {@literal key}. * diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java index f540c45dd..ab8c97702 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java @@ -143,9 +143,35 @@ public interface ReactiveRedisOperations { * @param pattern must not be {@literal null}. * @return * @see Redis Documentation: KEYS + * @deprecated Since 2.1. Use {@link #scan()} to iterate over the keyspace as {@link #keys(Object)} is a + * non-interruptible and expensive Redis operation. */ + @Deprecated Flux keys(K pattern); + /** + * Use a {@link Flux} to iterate over keys. The resulting {@link Flux} acts as a cursor and issues {@code SCAN} + * commands itself as long as the subscriber signals demand. + * + * @return never {@literal null}. + * @see Redis Documentation: SCAN + * @since 2.1 + */ + default Flux scan() { + return scan(ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over keys. The resulting {@link Flux} acts as a cursor and issues {@code SCAN} + * commands itself as long as the subscriber signals demand. + * + * @param options must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: SCAN + * @since 2.1 + */ + Flux scan(ScanOptions options); + /** * Return a random key from the keyspace. * diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index a88e62851..87d39b215 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -270,6 +270,19 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations scan(ScanOptions options) { + + Assert.notNull(options, "ScanOptions must not be null!"); + + return createFlux(connection -> connection.keyCommands().scan(options)) // + .map(this::readKey); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveRedisOperations#randomKey() diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveSetOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveSetOperations.java index 70641b83f..8ab9d5923 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveSetOperations.java @@ -234,6 +234,31 @@ public interface ReactiveSetOperations { */ Flux members(K key); + /** + * Use a {@link Flux} to iterate over entries in the set at {@code key} given {@link ScanOptions}. The resulting + * {@link Flux} acts as a cursor and issues {@code SSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: SSCAN + * @since 2.1 + */ + default Flux scan(K key) { + return scan(key, ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over entries in the set at {@code key} given {@link ScanOptions}. The resulting + * {@link Flux} acts as a cursor and issues {@code SSCAN} commands itself as long as the subscriber signals demand. + * + * @param key must not be {@literal null}. + * @param options must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: SSCAN + * @since 2.1 + */ + Flux scan(K key, ScanOptions options); + /** * Get random element from set at {@code key}. * diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java index f1262c96f..33092e4df 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java @@ -228,6 +228,33 @@ public interface ReactiveZSetOperations { */ Flux> reverseRangeByScoreWithScores(K key, Range range, Limit limit); + /** + * Use a {@link Flux} to iterate over entries in the sorted set at {@code key} given {@link ScanOptions}. The + * resulting {@link Flux} acts as a cursor and issues {@code ZSCAN} commands itself as long as the subscriber signals + * demand. + * + * @param key must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: ZSCAN + * @since 2.1 + */ + default Flux> scan(K key) { + return scan(key, ScanOptions.NONE); + } + + /** + * Use a {@link Flux} to iterate over entries in the sorted set at {@code key} given {@link ScanOptions}. The + * resulting {@link Flux} acts as a cursor and issues {@code ZSCAN} commands itself as long as the subscriber signals + * demand. + * + * @param key must not be {@literal null}. + * @param options must not be {@literal null}. + * @return never {@literal null}. + * @see Redis Documentation: ZSCAN + * @since 2.1 + */ + Flux> scan(K key, ScanOptions options); + /** * Count number of elements within sorted set with scores between {@code min} and {@code max}. * diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommandsTests.java index f5f5dbcec..839fcd780 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommandsTests.java @@ -28,6 +28,7 @@ import java.util.LinkedHashMap; import java.util.Map; import org.junit.Test; +import org.springframework.data.redis.core.ScanOptions; /** * @author Christoph Strobl @@ -233,6 +234,18 @@ public class LettuceReactiveHashCommandsTests extends LettuceReactiveCommandsTes .verifyComplete(); } + @Test // DATAREDIS-743 + public void hScanShouldIterateOverHash() { + + nativeCommands.hset(KEY_1, FIELD_1, VALUE_1); + nativeCommands.hset(KEY_1, FIELD_2, VALUE_2); + nativeCommands.hset(KEY_1, FIELD_3, VALUE_3); + + StepVerifier.create(connection.hashCommands().hScan(KEY_1_BBUFFER, ScanOptions.scanOptions().count(1).build())) // + .expectNextCount(3) // + .verifyComplete(); + } + @Test // DATAREDIS-698 public void hStrLenReturnsFieldLength() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommandsTests.java index 1ca41c6d0..44aad990c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommandsTests.java @@ -38,6 +38,7 @@ import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.data.redis.connection.ValueEncoding.RedisValueEncoding; +import org.springframework.data.redis.core.ScanOptions; import org.springframework.data.redis.test.util.MinimumRedisVersionRule; import org.springframework.test.annotation.IfProfileValue; @@ -95,6 +96,26 @@ public class LettuceReactiveKeyCommandsTests extends LettuceReactiveCommandsTest .expectNextCount(3).verifyComplete(); } + @Test // DATAREDIS-743 + public void scanShouldShouldIterateOverKeyspace() { + + nativeCommands.set(KEY_1, VALUE_2); + nativeCommands.set(KEY_2, VALUE_2); + nativeCommands.set(KEY_3, VALUE_3); + + nativeCommands.set(VALUE_1, KEY_1); + nativeCommands.set(VALUE_2, KEY_2); + nativeCommands.set(VALUE_3, KEY_3); + + StepVerifier.create(connection.keyCommands().scan(ScanOptions.scanOptions().count(2).build())) // + .expectNextCount(6) // + .verifyComplete(); + + StepVerifier.create(connection.keyCommands().scan(ScanOptions.scanOptions().count(2).match("key*").build())) // + .expectNextCount(3) // + .verifyComplete(); + } + @Test // DATAREDIS-525 public void randomKeyShouldReturnAnyKey() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommandsTests.java index 8330d027f..9484f97bf 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommandsTests.java @@ -27,6 +27,7 @@ import reactor.test.StepVerifier; import java.util.Arrays; import org.junit.Test; +import org.springframework.data.redis.core.ScanOptions; public class LettuceReactiveSetCommandsTests extends LettuceReactiveCommandsTestsBase { @@ -220,6 +221,21 @@ public class LettuceReactiveSetCommandsTests extends LettuceReactiveCommandsTest .verifyComplete(); } + @Test // DATAREDIS-743 + public void sScanShouldIterateOverSet() { + + nativeCommands.sadd(KEY_1, VALUE_1, VALUE_2, VALUE_3); + + StepVerifier.create(connection.setCommands().sScan(KEY_1_BBUFFER)) // + .expectNextCount(3) // + .verifyComplete(); + + StepVerifier + .create(connection.setCommands().sScan(KEY_1_BBUFFER, ScanOptions.scanOptions().match("value-3").build())) // + .expectNext(VALUE_3_BBUFFER) // + .verifyComplete(); + } + @Test // DATAREDIS-525 public void sRandMemberReturnsRandomMember() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java index f0fd0177a..79f3e3472 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java @@ -28,7 +28,7 @@ import org.hamcrest.collection.IsIterableContainingInOrder; import org.junit.Test; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.DefaultTuple; -import org.springframework.data.redis.test.util.LettuceRedisClientProvider; +import org.springframework.data.redis.core.ScanOptions; /** * @author Christoph Strobl @@ -276,6 +276,22 @@ public class LettuceReactiveZSetCommandsTests extends LettuceReactiveCommandsTes .verifyComplete(); } + @Test // DATAREDIS-743 + public void zScanShouldIterateOverSortedSet() { + + nativeCommands.zadd(KEY_1, 1D, VALUE_1); + nativeCommands.zadd(KEY_1, 2D, VALUE_2); + nativeCommands.zadd(KEY_1, 3D, VALUE_3); + + StepVerifier.create(connection.zSetCommands().zScan(KEY_1_BBUFFER, ScanOptions.scanOptions().count(1).build())) // + .expectNextCount(3).verifyComplete(); + + StepVerifier + .create(connection.zSetCommands().zScan(KEY_1_BBUFFER, ScanOptions.scanOptions().match(VALUE_2).build())) // + .expectNext(new DefaultTuple(VALUE_2_BBUFFER.array(), 2D)) // + .verifyComplete(); + } + @Test // DATAREDIS-525 public void zCountShouldCountValuesInRange() { diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveHashOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveHashOperationsIntegrationTests.java index a46926e05..078ffd19d 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveHashOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveHashOperationsIntegrationTests.java @@ -382,6 +382,31 @@ public class DefaultReactiveHashOperationsIntegrationTests { .verifyComplete(); } + @Test // DATAREDIS-743 + public void scan() { + + assumeTrue(hashKeyFactory instanceof StringObjectFactory && hashValueFactory instanceof StringObjectFactory); + + K key = keyFactory.instance(); + HK hashkey1 = hashKeyFactory.instance(); + HV hashvalue1 = hashValueFactory.instance(); + + HK hashkey2 = hashKeyFactory.instance(); + HV hashvalue2 = hashValueFactory.instance(); + + putAll(key, hashkey1, hashvalue1, hashkey2, hashvalue2); + + StepVerifier.create(hashOperations.scan(key).buffer(2)) // + .consumeNextWith(list -> { + + Entry entry1 = Collections.singletonMap(hashkey1, hashvalue1).entrySet().iterator().next(); + Entry entry2 = Collections.singletonMap(hashkey2, hashvalue2).entrySet().iterator().next(); + + assertThat(list).containsExactlyInAnyOrder(entry1, entry2); + }) // + .verifyComplete(); + } + @Test // DATAREDIS-602 public void delete() { diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveSetOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveSetOperationsIntegrationTests.java index aaa9e138a..ea6345f35 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveSetOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveSetOperationsIntegrationTests.java @@ -310,6 +310,25 @@ public class DefaultReactiveSetOperationsIntegrationTests { .consumeNextWith(actual -> assertThat(actual).isIn(value1, value2)).expectNextCount(1).verifyComplete(); } + @Test // DATAREDIS-743 + public void scan() { + + assumeFalse(valueFactory instanceof ByteBufferObjectFactory); + + K key = keyFactory.instance(); + V value1 = valueFactory.instance(); + V value2 = valueFactory.instance(); + + StepVerifier.create(setOperations.add(key, value1, value2)) // + .expectNext(2L) // + .verifyComplete(); + + StepVerifier.create(setOperations.scan(key)) // + .consumeNextWith(actual -> assertThat(actual).isIn(value1, value2)) // + .expectNextCount(1) // + .verifyComplete(); + } + @Test // DATAREDIS-602 public void randomMember() { diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java index d71930640..b85aeafda 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java @@ -15,6 +15,7 @@ */ package org.springframework.data.redis.core; +import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; import reactor.test.StepVerifier; @@ -389,6 +390,29 @@ public class DefaultReactiveZSetOperationsIntegrationTests { .verifyComplete(); } + @Test // DATAREDIS-743 + public void scan() { + + assumeFalse(valueFactory instanceof ByteBufferObjectFactory); + + K key = keyFactory.instance(); + V value1 = valueFactory.instance(); + V value2 = valueFactory.instance(); + + StepVerifier.create(zSetOperations.add(key, value1, 42.1)) // + .expectNext(true) // + .verifyComplete(); + StepVerifier.create(zSetOperations.add(key, value2, 10)) // + .expectNext(true) // + .verifyComplete(); + + StepVerifier.create(zSetOperations.scan(key)) // + .consumeNextWith(actual -> assertThat(actual).isIn(new DefaultTypedTuple<>(value1, 42.1), + new DefaultTypedTuple<>(value2, 10d))) // + .expectNextCount(1) // + .verifyComplete(); + } + @Test // DATAREDIS-602 public void count() { diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java index bff84e8ac..2a03d312e 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java @@ -25,6 +25,8 @@ import java.time.Duration; import java.time.Instant; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import org.junit.AfterClass; import org.junit.Before; @@ -116,6 +118,23 @@ public class ReactiveRedisTemplateIntegrationTests { StepVerifier.create(redisTemplate.hasKey(key)).expectNext(true).verifyComplete(); } + @Test // DATAREDIS-743 + public void scan() { + + assumeFalse(valueFactory.instance() instanceof Person); + + Map tuples = new HashMap<>(); + tuples.put(keyFactory.instance(), valueFactory.instance()); + tuples.put(keyFactory.instance(), valueFactory.instance()); + tuples.put(keyFactory.instance(), valueFactory.instance()); + + StepVerifier.create(redisTemplate.opsForValue().multiSet(tuples)).expectNext(true).verifyComplete(); + + StepVerifier.create(redisTemplate.scan().collectList()) // + .consumeNextWith(actual -> assertThat(actual).containsAll(tuples.keySet())) // + .verifyComplete(); + } + @Test // DATAREDIS-602 public void type() {