DATAREDIS-743 - Add Reactive SCAN, HSCAN, SSCAN, and ZSCAN support.

We now provide reactive SCAN support for keys, hashes, sets, and sorted sets. Scanning uses a cursor-based Flux and takes subscriber demand into account to issue the according scan commands once a batch of keys/values/entries was emitted. Specifying a SCAN limit controls the batch (page) size and can be used to optimize the number of roundtrips between the application and Redis.

Original Pull Request: #343

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<CommandResponse<KeyCommand, Flux<Map.Entry<ByteBuffer, ByteBuffer>>>> hGetAll(Publisher<KeyCommand> 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 <a href="http://redis.io/commands/hscan">Redis Documentation: HSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<Map.Entry<ByteBuffer, ByteBuffer>> 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 <a href="http://redis.io/commands/hscan">Redis Documentation: HSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<Map.Entry<ByteBuffer, ByteBuffer>> 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 <a href="http://redis.io/commands/hscan">Redis Documentation: HSCAN</a>
+	 * @since 2.1
+	 */
+	Flux<CommandResponse<KeyCommand, Flux<Map.Entry<ByteBuffer, ByteBuffer>>>> hScan(Publisher<KeyScanCommand> commands);
+
 	/**
 	 * @author Christoph Strobl
 	 * @see <a href="http://redis.io/commands/hstrlen">Redis Documentation: HSTRLEN</a>

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<MultiValueResponse<ByteBuffer, ByteBuffer>> keys(Publisher<ByteBuffer> 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 <a href="http://redis.io/commands/scan">Redis Documentation: SCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<ByteBuffer> 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 <a href="http://redis.io/commands/scan">Redis Documentation: SCAN</a>
+	 * @since 2.1
+	 */
+	Flux<ByteBuffer> scan(ScanOptions options);
+
 	/**
 	 * Return a random key from the keyspace.
 	 *

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
 	 */

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<CommandResponse<KeyCommand, Flux<ByteBuffer>>> sMembers(Publisher<KeyCommand> 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 <a href="http://redis.io/commands/sscan">Redis Documentation: SSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<ByteBuffer> 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 <a href="http://redis.io/commands/sscan">Redis Documentation: SSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<ByteBuffer> 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 <a href="http://redis.io/commands/sscan">Redis Documentation: SSCAN</a>
+	 * @since 2.1
+	 */
+	Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> sScan(Publisher<KeyScanCommand> commands);
+
 	/**
 	 * {@code SRANDMEMBER} command parameters.
 	 *

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<CommandResponse<ZRangeByScoreCommand, Flux<Tuple>>> zRangeByScore(Publisher<ZRangeByScoreCommand> 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 <a href="http://redis.io/commands/zscan">Redis Documentation: ZSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<Tuple> 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 <a href="http://redis.io/commands/zscan">Redis Documentation: ZSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<Tuple> 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 <a href="http://redis.io/commands/zscan">Redis Documentation: ZSCAN</a>
+	 * @since 2.1
+	 */
+	Flux<CommandResponse<KeyCommand, Flux<Tuple>>> zScan(Publisher<KeyScanCommand> commands);
+
 	/**
 	 * {@code ZCOUNT} command parameters.
 	 *

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java

@@ -204,7 +204,7 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
 						@Override
 						protected LettuceScanIteration<byte[]> doScan(io.lettuce.core.ScanCursor cursor, ScanOptions options) {
 
-							ScanArgs scanArgs = connection.getScanArgs(options);
+							ScanArgs scanArgs = LettuceConverters.toScanArgs(options);
 
 							KeyScanCursor<byte[]> keyScanCursor = client.scan(cursor, scanArgs);
 							return new LettuceScanIteration<>(keyScanCursor, keyScanCursor.getKeys());

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()) {

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}.
 	 *

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<byte[], byte[]> mapScanCursor = getConnection().hscan(key, scanCursor, scanArgs);
 				String nextCursorId = mapScanCursor.getCursor();

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<byte[]> keyScanCursor = getConnection().scan(cursor, scanArgs);
 				List<byte[]> keys = keyScanCursor.getKeys();

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<CommandResponse<KeyCommand, Flux<Map.Entry<ByteBuffer, ByteBuffer>>>> hScan(
+			Publisher<KeyScanCommand> 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<KeyValue<ByteBuffer, ByteBuffer>> result = ScanStream.hscan(cmd, command.getKey(),
+					LettuceConverters.toScanArgs(command.getOptions()));
+
+			Flux<Entry<ByteBuffer, ByteBuffer>> entryFlux = result.map(it -> new Entry<ByteBuffer, ByteBuffer>() {
+
+				@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)

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<ByteBuffer> 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()

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<CommandResponse<KeyCommand, Flux<ByteBuffer>>> sScan(Publisher<KeyScanCommand> 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<ByteBuffer> 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)

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<CommandResponse<KeyCommand, Flux<Tuple>>> zScan(Publisher<KeyScanCommand> 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<Tuple> 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)

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<byte[]> valueScanCursor = getConnection().sscan(key, scanCursor, scanArgs);
 				String nextCursorId = valueScanCursor.getCursor();

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<byte[]> scoredValueScanCursor = getConnection().zscan(key, scanCursor, scanArgs);
 				String nextCursorId = scoredValueScanCursor.getCursor();

src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java

@@ -221,6 +221,19 @@ class DefaultReactiveHashOperations<H, HK, HV> 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<Map.Entry<HK, HV>> 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)
 	 */

src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java

@@ -322,6 +322,18 @@ class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations<K, V>
 		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<V> 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)
 	 */

src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java

@@ -283,6 +283,18 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				connection -> 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<TypedTuple<V>> 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)
 	 */

src/main/java/org/springframework/data/redis/core/ReactiveHashOperations.java

@@ -146,6 +146,31 @@ public interface ReactiveHashOperations<H, HK, HV> {
 	 */
 	Flux<Map.Entry<HK, HV>> 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 <a href="http://redis.io/commands/hscan">Redis Documentation: HSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<Map.Entry<HK, HV>> 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 <a href="http://redis.io/commands/hscan">Redis Documentation: HSCAN</a>
+	 * @since 2.1
+	 */
+	Flux<Map.Entry<HK, HV>> scan(H key, ScanOptions options);
+
 	/**
 	 * Removes the given {@literal key}.
 	 *

src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java

@@ -143,9 +143,35 @@ public interface ReactiveRedisOperations<K, V> {
 	 * @param pattern must not be {@literal null}.
 	 * @return
 	 * @see <a href="http://redis.io/commands/keys">Redis Documentation: KEYS</a>
+	 * @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<K> 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 <a href="http://redis.io/commands/scan">Redis Documentation: SCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<K> 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 <a href="http://redis.io/commands/scan">Redis Documentation: SCAN</a>
+	 * @since 2.1
+	 */
+	Flux<K> scan(ScanOptions options);
+
 	/**
 	 * Return a random key from the keyspace.
 	 *

src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java

@@ -270,6 +270,19 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 				.map(this::readKey);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#scan(org.springframework.data.redis.core.ScanOptions)
+	 */
+	@Override
+	public Flux<K> 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()

src/main/java/org/springframework/data/redis/core/ReactiveSetOperations.java

@@ -234,6 +234,31 @@ public interface ReactiveSetOperations<K, V> {
 	 */
 	Flux<V> 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 <a href="http://redis.io/commands/sscan">Redis Documentation: SSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<V> 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 <a href="http://redis.io/commands/sscan">Redis Documentation: SSCAN</a>
+	 * @since 2.1
+	 */
+	Flux<V> scan(K key, ScanOptions options);
+
 	/**
 	 * Get random element from set at {@code key}.
 	 *

src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java

@@ -228,6 +228,33 @@ public interface ReactiveZSetOperations<K, V> {
 	 */
 	Flux<TypedTuple<V>> reverseRangeByScoreWithScores(K key, Range<Double> 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 <a href="http://redis.io/commands/zscan">Redis Documentation: ZSCAN</a>
+	 * @since 2.1
+	 */
+	default Flux<TypedTuple<V>> 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 <a href="http://redis.io/commands/zscan">Redis Documentation: ZSCAN</a>
+	 * @since 2.1
+	 */
+	Flux<TypedTuple<V>> scan(K key, ScanOptions options);
+
 	/**
 	 * Count number of elements within sorted set with scores between {@code min} and {@code max}.
 	 *