DATAREDIS-290 - Add support for SCAN command.

SCAN is currently only supported by jedis but can be emulated for
lettuce via eval.
Srp and JRedis will respond with UnsupportedOperationException.

We provide a Cursor implementation allowing to scroll through the
responses provided by the underlying connection. The cursor will fetch
additional results from redis server whenever its starting point has not
been reached and there are no more already loaded items available.

Only values returned by the last call to redis server are kept in memory
which allows scrolling through the entire collection without potentially
running into memory issues.

Original pull request: #71.

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

@@ -32,6 +32,8 @@ import org.springframework.data.redis.RedisSystemException;
 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.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
@@ -2215,6 +2217,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 		this.delegate.slaveOfNoOne();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions)
+	 */
+	@Override
+	public Cursor<byte[]> scan(ScanOptions options) {
+		return this.delegate.scan(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

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

@@ -18,6 +18,9 @@ package org.springframework.data.redis.connection;
 import java.util.List;
 import java.util.Set;
 
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+
 /**
  * Key-specific commands supported by Redis.
  * 
@@ -62,6 +65,17 @@ public interface RedisKeyCommands {
 	 */
 	Set<byte[]> keys(byte[] pattern);
 
+	/**
+	 * Use a {@link Cursor} to iterate over keys. <br />
+	 * 
+	 * @see http://redis.io/commands/scan
+	 * @param count
+	 * @param pattern
+	 * @return
+	 * @since 1.4
+	 */
+	Cursor<byte[]> scan(ScanOptions options);
+
 	/**
 	 * Return a random key from the keyspace.
 	 * 

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

@@ -42,10 +42,15 @@ import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.Subscription;
 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.ScanCursor;
+import org.springframework.data.redis.core.ScanIteration;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
 
 import redis.clients.jedis.BinaryJedis;
 import redis.clients.jedis.BinaryJedisPubSub;
@@ -58,6 +63,7 @@ import redis.clients.jedis.Protocol;
 import redis.clients.jedis.Protocol.Command;
 import redis.clients.jedis.Queable;
 import redis.clients.jedis.Response;
+import redis.clients.jedis.ScanParams;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.Transaction;
 import redis.clients.jedis.ZParams;
@@ -2900,6 +2906,47 @@ public class JedisConnection implements RedisConnection {
 		}
 	}
 
+	public Cursor<byte[]> scan() {
+		return scan(ScanOptions.NONE);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions)
+	 */
+	public Cursor<byte[]> scan(ScanOptions options) {
+		return scan(0, options != null ? options : ScanOptions.NONE);
+	}
+
+	public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
+
+		return new ScanCursor<byte[]>(cursorId, options) {
+
+			@Override
+			protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+
+				if (isQueueing() || isPipelined()) {
+					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());
+					}
+				}
+
+				redis.clients.jedis.ScanResult<String> result = jedis.scan(Long.toString(cursorId), sp);
+				return new ScanIteration<byte[]>(Long.valueOf(result.getStringCursor()), JedisConverters.stringListToByteList()
+						.convert(result.getResult()));
+			}
+		}.open();
+
+	}
+
 	/**
 	 * Specifies if pipelined results should be converted to the expected data type. If false, results of
 	 * {@link #closePipeline()} and {@link #exec()} will be of the type returned by the Jedis driver

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

@@ -43,6 +43,8 @@ import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.Subscription;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -1236,4 +1238,13 @@ public class JredisConnection implements RedisConnection {
 			throw convertJredisAccessException(e);
 		}
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions)
+	 */
+	@Override
+	public Cursor<byte[]> scan(ScanOptions options) {
+		throw new UnsupportedOperationException("'SCAN' command is not supported for jredis.");
+	}
 }

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

@@ -49,11 +49,16 @@ import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.Subscription;
 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.RedisCommand;
+import org.springframework.data.redis.core.ScanCursor;
+import org.springframework.data.redis.core.ScanIteration;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 import com.lambdaworks.redis.RedisAsyncConnection;
 import com.lambdaworks.redis.RedisClient;
@@ -3027,6 +3032,51 @@ public class LettuceConnection implements RedisConnection {
 		}
 	}
 
+	public Cursor<byte[]> scan() {
+		return scan(0, ScanOptions.NONE);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions)
+	 */
+	public Cursor<byte[]> scan(ScanOptions options) {
+		return scan(0, options != null ? options : ScanOptions.NONE);
+	}
+
+	public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
+
+		return new ScanCursor<byte[]>(cursorId, options) {
+
+			@SuppressWarnings("unchecked")
+			@Override
+			protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+
+				if (isQueueing() || isPipelined()) {
+					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 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();
+
+	}
+
 	/**
 	 * Specifies if pipelined and transaction results should be converted to the expected data type. If false, results of
 	 * {@link #closePipeline()} and {@link #exec()} will be of the type returned by the Lettuce driver

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

@@ -40,6 +40,8 @@ import org.springframework.data.redis.connection.RedisSubscribedConnectionExcept
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.Subscription;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.util.Assert;
 
@@ -2331,6 +2333,15 @@ public class SrpConnection implements RedisConnection {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions)
+	 */
+	@Override
+	public Cursor<byte[]> scan(ScanOptions options) {
+		throw new UnsupportedOperationException("'SCAN' 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/Cursor.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+import java.io.Closeable;
+import java.util.Iterator;
+
+/**
+ * @author Christoph Strobl
+ * @param <T>
+ * @since 1.4
+ */
+public interface Cursor<T> extends Iterator<T>, Closeable {
+
+	/**
+	 * Get the reference cursor. <br />
+	 * <strong>NOTE:</strong> the id might change while iterating items.
+	 * 
+	 * @return
+	 */
+	long getCursorId();
+
+	/**
+	 * @return Returns true if cursor closed.
+	 */
+	boolean isClosed();
+
+	/**
+	 * Opens cursor and returns itself.
+	 * 
+	 * @return
+	 */
+	Cursor<T> open();
+
+	/**
+	 * @return Returns the current position of the cursor.
+	 */
+	long getPosition();
+
+}

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

@@ -0,0 +1,270 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Redis client agnostic {@link Cursor} implementation continuously loading additional results from Redis server until
+ * reaching its starting point {@code zero}. <br />
+ * <strong>Note:</strong> Please note that the {@link ScanCursor} has to be initialized ({@link #open()} prior to usage.
+ * 
+ * @author Christoph Strobl
+ * @author Thomas Darimont
+ * @param <T>
+ * @since 1.4
+ */
+public abstract class ScanCursor<T> implements Cursor<T> {
+
+	private CursorState state;
+	private long cursorId;
+	private Iterator<T> delegate;
+	private final ScanOptions scanOptions;
+	private long position;
+
+	/**
+	 * Crates new {@link ScanCursor} with {@code id=0} and {@link ScanOptions#NONE}
+	 */
+	public ScanCursor() {
+		this(ScanOptions.NONE);
+	}
+
+	/**
+	 * Crates new {@link ScanCursor} with {@code id=0}.
+	 * 
+	 * @param options
+	 */
+	public ScanCursor(ScanOptions options) {
+		this(0, options);
+	}
+
+	/**
+	 * Crates new {@link ScanCursor} with {@link ScanOptions#NONE}
+	 * 
+	 * @param cursorId
+	 */
+	public ScanCursor(long cursorId) {
+		this(cursorId, ScanOptions.NONE);
+	}
+
+	/**
+	 * Crates new {@link ScanCursor}
+	 * 
+	 * @param cursorId
+	 * @param options Defaulted to {@link ScanOptions#NONE} if nulled.
+	 */
+	public ScanCursor(long cursorId, ScanOptions options) {
+
+		this.scanOptions = options != null ? options : ScanOptions.NONE;
+		this.cursorId = cursorId;
+		this.state = CursorState.CLOSED;
+		this.delegate = Collections.emptyIterator();
+	}
+
+	private void scan(long cursorId) {
+
+		ScanIteration<T> result = doScan(cursorId, this.scanOptions);
+		processScanResult(result);
+	}
+
+	/**
+	 * Performs the actual scan command using the native client implementation. The given {@literal options} are never
+	 * {@code null}.
+	 * 
+	 * @param cursorId
+	 * @param options
+	 * @return
+	 */
+	protected abstract ScanIteration<T> doScan(long cursorId, ScanOptions options);
+
+	/**
+	 * Initialize the {@link Cursor} prior to usage.
+	 */
+	public final ScanCursor<T> open() {
+
+		if (isClosed()) {
+			doOpen(cursorId);
+			state = CursorState.OPEN;
+		}
+
+		return this;
+	}
+
+	/**
+	 * Customization hook when calling {@link #open()}.
+	 * 
+	 * @param cursorId
+	 */
+	protected void doOpen(long cursorId) {
+		scan(cursorId);
+	}
+
+	private void processScanResult(ScanIteration<T> result) {
+
+		if (result == null) {
+
+			resetDelegate();
+			state = CursorState.FINISHED;
+			return;
+		}
+
+		cursorId = Long.valueOf(result.getCursorId());
+
+		if (cursorId == 0) {
+			state = CursorState.FINISHED;
+		}
+
+		if (!CollectionUtils.isEmpty(result.getItems())) {
+			delegate = result.iterator();
+		} else {
+			resetDelegate();
+		}
+	}
+
+	private void resetDelegate() {
+		delegate = Collections.emptyIterator();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.Cursor#getCursorId()
+	 */
+	@Override
+	public long getCursorId() {
+		return cursorId;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.util.Iterator#hasNext()
+	 */
+	@Override
+	public boolean hasNext() {
+
+		assertCursorIsOpen();
+
+		if (delegate.hasNext()) {
+			return true;
+		}
+
+		if (cursorId > 0) {
+			return true;
+		}
+
+		return false;
+	}
+
+	private void assertCursorIsOpen() {
+
+		if (isClosed()) {
+			throw new InvalidDataAccessApiUsageException("Cannot access closed cursor. Did you forget to call open()?");
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.util.Iterator#next()
+	 */
+	@Override
+	public T next() {
+
+		assertCursorIsOpen();
+
+		if (state != CursorState.FINISHED && !delegate.hasNext()) {
+			scan(cursorId);
+		}
+
+		if (!delegate.hasNext()) {
+			throw new NoSuchElementException("No more elements available for cursor " + cursorId + ".");
+		}
+
+		T next = moveNext(delegate);
+		position++;
+
+		return next;
+	}
+
+	/**
+	 * Fetch the next item from the underlying {@link Iterable}.
+	 * 
+	 * @param source
+	 * @return
+	 */
+	protected T moveNext(Iterator<T> source) {
+		return source.next();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.util.Iterator#remove()
+	 */
+	@Override
+	public void remove() {
+		throw new UnsupportedOperationException("Remove is not supported");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.io.Closeable#close()
+	 */
+	@Override
+	public final void close() throws IOException {
+
+		try {
+			doClose();
+		}finally {
+			state = CursorState.CLOSED;
+		}
+	}
+
+	/**
+	 * Customization hook for cleaning up resources on when calling {@link #close()}.
+	 */
+	protected void doClose() {
+		resetDelegate();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.Cursor#isClosed()
+	 */
+	@Override
+	public boolean isClosed() {
+		return state == CursorState.CLOSED;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.Cursor#getPosition()
+	 */
+	@Override
+	public long getPosition() {
+		return position;
+	}
+
+	/**
+	 * @author Thomas Darimont
+	 */
+	enum CursorState{
+		OPEN, FINISHED, CLOSED;
+	}
+}

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * {@link ScanIteration} holds the values contained in Redis {@literal Multibulk reply} on exectuting {@literal SCAN}
+ * command.
+ * 
+ * @author Christoph Strobl
+ * @since 1.4
+ */
+public class ScanIteration<T> implements Iterable<T> {
+
+	private final long cursorId;
+	private final List<T> items;
+
+	/**
+	 * @param cursorId
+	 * @param items
+	 */
+	public ScanIteration(long cursorId, List<T> items) {
+
+		this.cursorId = cursorId;
+		this.items = (items != null ? new ArrayList<T>(items) : Collections.<T> emptyList());
+	}
+
+	/**
+	 * The cursor id to be used for subsequent requests.
+	 * 
+	 * @return
+	 */
+	public long getCursorId() {
+		return cursorId;
+	}
+
+	/**
+	 * Get the items returned.
+	 * 
+	 * @return
+	 */
+	public List<T> getItems() {
+		return items;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Iterable#iterator()
+	 */
+	@Override
+	public Iterator<T> iterator() {
+		return items.iterator();
+	}
+
+}

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

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+/**
+ * Options to be used for with {@literal SCAN} command.
+ * 
+ * @author Christoph Strobl
+ * @author Thomas Darimont
+ * @since 1.4
+ */
+public class ScanOptions {
+
+	public static ScanOptions NONE = new ScanOptions();
+
+	private Long count;
+	private String pattern;
+
+	private ScanOptions() {
+	}
+
+	/**
+	 * Static factory method that returns a new {@link ScanOptionsBuilder}.
+	 *
+	 * @return
+	 */
+	public static ScanOptionsBuilder scanOptions(){
+		return new ScanOptionsBuilder();
+	}
+
+	public Long getCount() {
+		return count;
+	}
+
+	public String getPattern() {
+		return pattern;
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.4
+	 */
+	public static class ScanOptionsBuilder {
+
+		ScanOptions options;
+
+		public ScanOptionsBuilder() {
+			options = new ScanOptions();
+		}
+
+		/**
+		 * Returns the current {@link ScanOptionsBuilder} configured with the given {@code count}.
+		 *
+		 * @param count
+		 * @return
+		 */
+		public ScanOptionsBuilder count(long count) {
+			options.count = count;
+			return this;
+		}
+
+		/**
+		 * Returns the current {@link ScanOptionsBuilder} configured with the given {@code pattern}.
+		 *
+		 * @param pattern
+		 * @return
+		 */
+		public ScanOptionsBuilder match(String pattern) {
+			options.pattern = pattern;
+			return this;
+		}
+
+		public ScanOptions build() {
+			return options;
+		}
+
+	}
+
+}