DATAREDIS-304 - Add support for SSCAN command.

SSCAN command is natively supported by jedis and can be emulated for lettuce. JRedis and SRP will throw UnsupportedOperationException. sScan is available on RedisConneciton and RedisSetOperations returning basically a Cursor that allows iteration over the defined values for a given key. Currently the jedis driver does not directly expose the binary version of sscan which leads to invalid results when using a non String compatible converter along with the RedisTemplate. We’ll change this as soon as a newer version of jedis is available. Along the way loading behavior of ScanCursor has been fixed, preventing it from indicating next values available when there effectively are no more. Orignal pull request: spring-projects#76.

Author: christophstrobl

src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java

@@ -32,6 +32,7 @@
 import org.springframework.data.redis.connection.convert.ListConverter;
 import org.springframework.data.redis.connection.convert.MapConverter;
 import org.springframework.data.redis.connection.convert.SetConverter;
+import org.springframework.data.redis.core.ConvertingCursor;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.RedisClientInfo;
@@ -2226,6 +2227,15 @@ public Cursor<byte[]> scan(ScanOptions options) {
 return this.delegate.scan(options);
 }
 
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisSetCommands#scan(byte[], org.springframework.data.redis.core.ScanOptions)
+ */
+@Override
+public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
+return this.delegate.sScan(key, options);
+}
+
 /**
  * Specifies if pipelined and tx results should be deserialized to Strings. If false, results of
  * {@link #closePipeline()} and {@link #exec()} will be of the type returned by the underlying connection
@@ -2299,4 +2309,23 @@ public void killClient(String host, int port) {
 public String getClientName() {
 return this.delegate.getClientName();
 }
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.StringRedisConnection#sScan(java.lang.String, org.springframework.data.redis.core.ScanOptions)
+ */
+@Override
+public Cursor<String> sScan(String key, ScanOptions options) {
+
+return new ConvertingCursor<byte[], String>(this.delegate.sScan(this.serialize(key), options),
+new Converter<byte[], String>() {
+
+@Override
+public String convert(byte[] source) {
+return serializer.deserialize(source);
+}
+});
+
+}
+
 }

src/main/java/org/springframework/data/redis/connection/RedisSetCommands.java

@@ -18,6 +18,9 @@
 import java.util.List;
 import java.util.Set;
 
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+
 /**
  * Set-specific commands supported by Redis.
  * 
@@ -166,4 +169,15 @@ public interface RedisSetCommands {
  * @return
  */
 List<byte[]> sRandMember(byte[] key, long count);
+
+/**
+ * Use a {@link Cursor} to iterate over elements in set at {@code key}.
+ * 
+ * @since 1.4
+ * @see http://redis.io/commands/scan
+ * @param key
+ * @param options
+ * @return
+ */
+Cursor<byte[]> sScan(byte[] key, ScanOptions options);
 }

src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java

@@ -20,7 +20,9 @@
 import java.util.Map;
 import java.util.Set;
 
+import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.data.redis.serializer.RedisSerializer;
@@ -316,4 +318,13 @@ public interface StringTuple extends Tuple {
  * @since 1.3
  */
 List<RedisClientInfo> getClientList();
+
+/**
+ * @see RedisSetCommands#sScan(byte[], ScanOptions)
+ * @param key
+ * @param options
+ * @return
+ * @since 1.4
+ */
+Cursor<String> sScan(String key, ScanOptions options);
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -43,6 +43,7 @@
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.connection.convert.TransactionResultConverter;
 import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -2918,6 +2919,12 @@ public Cursor<byte[]> scan(ScanOptions options) {
 return scan(0, options != null ? options : ScanOptions.NONE);
 }
 
+/**
+ * @since 1.4
+ * @param cursorId
+ * @param options
+ * @return
+ */
 public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
 
 return new ScanCursor<byte[]>(cursorId, options) {
@@ -2929,22 +2936,65 @@ protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
 throw new UnsupportedOperationException("'SCAN' cannot be called in pipeline / transaction mode.");
 }
 
-ScanParams sp = new ScanParams();
-if (!options.equals(ScanOptions.NONE)) {
-if (options.getCount() != null) {
-sp.count(options.getCount().intValue());
-}
-if (StringUtils.hasText(options.getPattern())) {
-sp.match(options.getPattern());
-}
+ScanParams params = prepareScanParams(options);
+redis.clients.jedis.ScanResult<String> result = jedis.scan(Long.toString(cursorId), params);
+return new ScanIteration<byte[]>(Long.valueOf(result.getStringCursor()), JedisConverters.stringListToByteList()
+.convert(result.getResult()));
+}
+
+}.open();
+
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
+ */
+@Override
+public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
+return sScan(key, 0, options);
+}
+
+/**
+ * @since 1.4
+ * @param key
+ * @param cursorId
+ * @param options
+ * @return
+ */
+public Cursor<byte[]> sScan(byte[] key, long cursorId, ScanOptions options) {
+
+return new KeyBoundCursor<byte[]>(key, cursorId, options) {
+
+@Override
+protected ScanIteration<byte[]> doScan(byte[] key, long cursorId, ScanOptions options) {
+
+if (isQueueing() || isPipelined()) {
+throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
 }
 
-redis.clients.jedis.ScanResult<String> result = jedis.scan(Long.toString(cursorId), sp);
+ScanParams params = prepareScanParams(options);
+
+// TODO: use binary version of jedis.sscan (in v.2.4.3) to avoid potentially invalid representations.
+redis.clients.jedis.ScanResult<String> result = jedis.sscan(JedisConverters.toString(key),
+Long.toString(cursorId), params);
 return new ScanIteration<byte[]>(Long.valueOf(result.getStringCursor()), JedisConverters.stringListToByteList()
 .convert(result.getResult()));
 }
 }.open();
+}
 
+private ScanParams prepareScanParams(ScanOptions options) {
+ScanParams sp = new ScanParams();
+if (!options.equals(ScanOptions.NONE)) {
+if (options.getCount() != null) {
+sp.count(options.getCount().intValue());
+}
+if (StringUtils.hasText(options.getPattern())) {
+sp.match(options.getPattern());
+}
+}
+return sp;
 }
 
 /**

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java

@@ -1247,4 +1247,13 @@ public void slaveOfNoOne() {
 public Cursor<byte[]> scan(ScanOptions options) {
 throw new UnsupportedOperationException("'SCAN' command is not supported for jredis.");
 }
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
+ */
+@Override
+public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
+throw new UnsupportedOperationException("'SSCAN' command is not uspported for jredis");
+}
 }

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

@@ -50,6 +50,7 @@
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.connection.convert.TransactionResultConverter;
 import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.RedisCommand;
 import org.springframework.data.redis.core.ScanCursor;
 import org.springframework.data.redis.core.ScanIteration;
@@ -3044,6 +3045,12 @@ public Cursor<byte[]> scan(ScanOptions options) {
 return scan(0, options != null ? options : ScanOptions.NONE);
 }
 
+/**
+ * @since 1.4
+ * @param cursorId
+ * @param options
+ * @return
+ */
 public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
 
 return new ScanCursor<byte[]>(cursorId, options) {
@@ -3056,25 +3063,70 @@ protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
 throw new UnsupportedOperationException("'SCAN' cannot be called in pipeline / transaction mode.");
 }
 
-String params = " ," + cursorId;
-if (!options.equals(ScanOptions.NONE)) {
-if (options.getCount() != null) {
-params += (", 'count', " + options.getCount());
-}
-if (StringUtils.hasText(options.getPattern())) {
-params += (", 'match' , '" + options.getPattern() + "'");
-}
+String params = " ," + cursorId + prepareScanParams(options);
+String script = "return redis.call('SCAN'" + params + ")";
+
+List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
+String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
+
+return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), ((ArrayList<byte[]>) result.get(1)));
+}
+}.open();
+
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
+ */
+@Override
+public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
+return sScan(key, 0, options);
+}
+
+/**
+ * @since 1.4
+ * @param key
+ * @param cursorId
+ * @param options
+ * @return
+ */
+public Cursor<byte[]> sScan(byte[] key, long cursorId, ScanOptions options) {
+
+return new KeyBoundCursor<byte[]>(key, cursorId, options) {
+
+@SuppressWarnings("unchecked")
+@Override
+protected ScanIteration<byte[]> doScan(byte[] key, long cursorId, ScanOptions options) {
+
+if (isQueueing() || isPipelined()) {
+throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
 }
 
-String script = "return redis.call('SCAN'" + params + ")";
+String params = " ,'" + LettuceConverters.bytesToString().convert(key) + "', " + cursorId
++ prepareScanParams(options);
+String script = "return redis.call('SSCAN'" + params + ")";
 
 List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
 String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
 
 return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), ((ArrayList<byte[]>) result.get(1)));
 }
 }.open();
+}
 
+private String prepareScanParams(ScanOptions options) {
+
+String params = "";
+if (!options.equals(ScanOptions.NONE)) {
+if (options.getCount() != null) {
+params += (", 'count', " + options.getCount());
+}
+if (StringUtils.hasText(options.getPattern())) {
+params += (", 'match' , '" + options.getPattern() + "'");
+}
+}
+return params;
 }
 
 /**

src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java

@@ -2342,6 +2342,15 @@ public Cursor<byte[]> scan(ScanOptions options) {
 throw new UnsupportedOperationException("'SCAN' command is not supported for Srp.");
 }
 
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
+ */
+@Override
+public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
+throw new UnsupportedOperationException("'SSCAN' command is not supported for Srp.");
+}
+
 private List<Object> closeTransaction() {
 List<Object> results = Collections.emptyList();
 if (txTracker != null) {

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

@@ -17,6 +17,7 @@
 package org.springframework.data.redis.core;
 
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 
@@ -72,4 +73,11 @@ public interface BoundSetOperations<K, V> extends BoundKeyOperations<K> {
 V pop();
 
 Long size();
+
+/**
+ * @param options
+ * @return
+ * @since 1.4
+ */
+Iterator<V> sScan(ScanOptions options);
 }