From d427928386b67bfdc383610fdbb7d623839aa5a7 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Wed, 30 Apr 2014 14:12:56 +0200 Subject: [PATCH] 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. --- .../DefaultStringRedisConnection.java | 11 + .../redis/connection/RedisKeyCommands.java | 14 + .../connection/jedis/JedisConnection.java | 47 +++ .../connection/jredis/JredisConnection.java | 11 + .../connection/lettuce/LettuceConnection.java | 50 ++++ .../redis/connection/srp/SrpConnection.java | 11 + .../data/redis/core/Cursor.java | 53 ++++ .../data/redis/core/ScanCursor.java | 270 ++++++++++++++++++ .../data/redis/core/ScanIteration.java | 72 +++++ .../data/redis/core/ScanOptions.java | 92 ++++++ .../AbstractConnectionIntegrationTests.java | 37 +++ .../data/redis/core/ScanCursorUnitTests.java | 208 ++++++++++++++ 12 files changed, 876 insertions(+) create mode 100644 src/main/java/org/springframework/data/redis/core/Cursor.java create mode 100644 src/main/java/org/springframework/data/redis/core/ScanCursor.java create mode 100644 src/main/java/org/springframework/data/redis/core/ScanIteration.java create mode 100644 src/main/java/org/springframework/data/redis/core/ScanOptions.java create mode 100644 src/test/java/org/springframework/data/redis/core/ScanCursorUnitTests.java diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index 0e3c471bf..733e80a65 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/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 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 diff --git a/src/main/java/org/springframework/data/redis/connection/RedisKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisKeyCommands.java index 84506960a..37762d4d1 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisKeyCommands.java +++ b/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 keys(byte[] pattern); + /** + * Use a {@link Cursor} to iterate over keys.
+ * + * @see http://redis.io/commands/scan + * @param count + * @param pattern + * @return + * @since 1.4 + */ + Cursor scan(ScanOptions options); + /** * Return a random key from the keyspace. * diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 44f22a4e3..ae3fc2a42 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/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 scan() { + return scan(ScanOptions.NONE); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions) + */ + public Cursor scan(ScanOptions options) { + return scan(0, options != null ? options : ScanOptions.NONE); + } + + public Cursor scan(long cursorId, ScanOptions options) { + + return new ScanCursor(cursorId, options) { + + @Override + protected ScanIteration 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 result = jedis.scan(Long.toString(cursorId), sp); + return new ScanIteration(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 diff --git a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java index 194a07e0a..1c797b4a7 100644 --- a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java +++ b/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 scan(ScanOptions options) { + throw new UnsupportedOperationException("'SCAN' command is not supported for jredis."); + } } 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 9fb72fe7a..89c19b479 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 @@ -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 scan() { + return scan(0, ScanOptions.NONE); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisKeyCommands#scan(org.springframework.data.redis.core.ScanOptions) + */ + public Cursor scan(ScanOptions options) { + return scan(0, options != null ? options : ScanOptions.NONE); + } + + public Cursor scan(long cursorId, ScanOptions options) { + + return new ScanCursor(cursorId, options) { + + @SuppressWarnings("unchecked") + @Override + protected ScanIteration 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(Long.valueOf(nextCursorId), ((ArrayList) 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 diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index be9aa3af5..1babf2f3f 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/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 scan(ScanOptions options) { + throw new UnsupportedOperationException("'SCAN' command is not supported for Srp."); + } + private List closeTransaction() { List results = Collections.emptyList(); if (txTracker != null) { diff --git a/src/main/java/org/springframework/data/redis/core/Cursor.java b/src/main/java/org/springframework/data/redis/core/Cursor.java new file mode 100644 index 000000000..12c3f3b2a --- /dev/null +++ b/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 + * @since 1.4 + */ +public interface Cursor extends Iterator, Closeable { + + /** + * Get the reference cursor.
+ * NOTE: 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 open(); + + /** + * @return Returns the current position of the cursor. + */ + long getPosition(); + +} diff --git a/src/main/java/org/springframework/data/redis/core/ScanCursor.java b/src/main/java/org/springframework/data/redis/core/ScanCursor.java new file mode 100644 index 000000000..e26e021d3 --- /dev/null +++ b/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}.
+ * Note: Please note that the {@link ScanCursor} has to be initialized ({@link #open()} prior to usage. + * + * @author Christoph Strobl + * @author Thomas Darimont + * @param + * @since 1.4 + */ +public abstract class ScanCursor implements Cursor { + + private CursorState state; + private long cursorId; + private Iterator 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 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 doScan(long cursorId, ScanOptions options); + + /** + * Initialize the {@link Cursor} prior to usage. + */ + public final ScanCursor 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 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 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; + } +} diff --git a/src/main/java/org/springframework/data/redis/core/ScanIteration.java b/src/main/java/org/springframework/data/redis/core/ScanIteration.java new file mode 100644 index 000000000..77b6edd5e --- /dev/null +++ b/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 implements Iterable { + + private final long cursorId; + private final List items; + + /** + * @param cursorId + * @param items + */ + public ScanIteration(long cursorId, List items) { + + this.cursorId = cursorId; + this.items = (items != null ? new ArrayList(items) : Collections. emptyList()); + } + + /** + * The cursor id to be used for subsequent requests. + * + * @return + */ + public long getCursorId() { + return cursorId; + } + + /** + * Get the items returned. + * + * @return + */ + public List getItems() { + return items; + } + + /* + * (non-Javadoc) + * @see java.lang.Iterable#iterator() + */ + @Override + public Iterator iterator() { + return items.iterator(); + } + +} diff --git a/src/main/java/org/springframework/data/redis/core/ScanOptions.java b/src/main/java/org/springframework/data/redis/core/ScanOptions.java new file mode 100644 index 000000000..357f429cf --- /dev/null +++ b/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; + } + + } + +} diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 37e1a19e3..40d0368bf 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -19,6 +19,7 @@ import static org.hamcrest.CoreMatchers.*; import static org.junit.Assert.*; import static org.junit.Assume.*; import static org.springframework.data.redis.SpinBarrier.*; +import static org.springframework.data.redis.core.ScanOptions.*; import java.util.ArrayList; import java.util.Arrays; @@ -41,6 +42,7 @@ import org.hamcrest.core.IsNot; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.junit.internal.AssumptionViolatedException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.RedisSystemException; @@ -53,6 +55,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; import org.springframework.data.redis.connection.SortParameters.Order; import org.springframework.data.redis.connection.StringRedisConnection.StringTuple; +import org.springframework.data.redis.core.Cursor; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.types.RedisClientInfo; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; @@ -1913,6 +1916,40 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(info.getDatabaseId(), is(notNullValue())); } + /** + * @see DATAREDIS-290 + */ + @Test + @IfProfileValue(name = "redisVersion", value = "2.8") + public void scanShouldReadEntireValueRange() { + + if (!ConnectionUtils.isJedis(connectionFactory) && !ConnectionUtils.isLettuce(connectionFactory)) { + throw new AssumptionViolatedException("SCAN is only available for jedis and lettuce"); + } + + if (connection.isPipelined() || connection.isQueueing()) { + throw new AssumptionViolatedException("SCAN is only available in non pipeline | queue mode."); + } + + connection.set("spring", "data"); + + int itemCount = 22; + for (int i = 0; i < itemCount; i++) { + connection.set(("key_" + i), ("foo_" + i)); + } + + Cursor cursor = connection.scan(scanOptions().count(20).match("ke*").build()); + + int i = 0; + while (cursor.hasNext()) { + byte[] value = cursor.next(); + assertThat(new String(value), not(containsString("spring"))); + i++; + } + + assertThat(i, is(itemCount)); + } + protected void verifyResults(List expected) { assertEquals(expected, getResults()); } diff --git a/src/test/java/org/springframework/data/redis/core/ScanCursorUnitTests.java b/src/test/java/org/springframework/data/redis/core/ScanCursorUnitTests.java new file mode 100644 index 000000000..d2a8d1295 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/ScanCursorUnitTests.java @@ -0,0 +1,208 @@ +/* + * 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 static org.hamcrest.core.Is.*; +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.Arrays; +import java.util.LinkedList; +import java.util.NoSuchElementException; +import java.util.Queue; +import java.util.Stack; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.springframework.dao.InvalidDataAccessApiUsageException; + +/** + * @author Christoph Strobl + */ +public class ScanCursorUnitTests { + + public @Rule ExpectedException exception = ExpectedException.none(); + + /** + * @see DATAREDIS-290 + */ + @Test + public void cursorShouldNotLoopWhenNoValuesFound() { + + CapturingCursorDummy cursor = initCursor(new LinkedList>()); + assertThat(cursor.hasNext(), is(false)); + } + + /** + * @see DATAREDIS-290 + */ + @Test(expected = NoSuchElementException.class) + public void cursorShouldReturnNullWhenNoNextElementAvailable() { + initCursor(new LinkedList>()).next(); + } + + /** + * @see DATAREDIS-290 + */ + @Test + public void cursorShouldNotLoopWhenReachingStartingPointInFistLoop() { + + LinkedList> values = new LinkedList>(); + values.add(createIteration(0, "spring", "data", "redis")); + CapturingCursorDummy cursor = initCursor(values); + + assertThat(cursor.next(), is("spring")); + assertThat(cursor.getCursorId(), is(0L)); + assertThat(cursor.hasNext(), is(true)); + + assertThat(cursor.next(), is("data")); + assertThat(cursor.getCursorId(), is(0L)); + assertThat(cursor.hasNext(), is(true)); + + assertThat(cursor.next(), is("redis")); + assertThat(cursor.getCursorId(), is(0L)); + assertThat(cursor.hasNext(), is(false)); + } + + /** + * @see DATAREDIS-290 + */ + @Test + public void cursorShouldStopLoopWhenReachingStartingPoint() { + + LinkedList> values = new LinkedList>(); + values.add(createIteration(1, "spring")); + values.add(createIteration(2, "data")); + values.add(createIteration(0, "redis")); + CapturingCursorDummy cursor = initCursor(values); + + assertThat(cursor.next(), is("spring")); + assertThat(cursor.getCursorId(), is(1L)); + assertThat(cursor.hasNext(), is(true)); + + assertThat(cursor.next(), is("data")); + assertThat(cursor.getCursorId(), is(2L)); + assertThat(cursor.hasNext(), is(true)); + + assertThat(cursor.next(), is("redis")); + assertThat(cursor.getCursorId(), is(0L)); + assertThat(cursor.hasNext(), is(false)); + } + + /** + * @see DATAREDIS-290 + */ + @Test + public void shouldThrowExceptionWhenAccessingClosedCursor() { + + CapturingCursorDummy cursor = new CapturingCursorDummy(null); + + assertThat(cursor.isClosed(), is(true)); + + exception.expect(InvalidDataAccessApiUsageException.class); + exception.expectMessage("closed cursor"); + + cursor.next(); + } + + /** + * @see DATAREDIS-290 + */ + @Test + public void repoeningCursorShouldHappenAtLastPosition() throws IOException { + + LinkedList> values = new LinkedList>(); + values.add(createIteration(1, "spring")); + values.add(createIteration(2, "data")); + values.add(createIteration(0, "redis")); + Cursor cursor = initCursor(values).open(); + + cursor.open(); + + assertThat(cursor.next(), is("spring")); + assertThat(cursor.getCursorId(), is(1L)); + assertThat(cursor.hasNext(), is(true)); + + // close the cursor + cursor.close(); + assertThat(cursor.isClosed(), is(true)); + + // reopen cursor at last position + cursor.open(); + + assertThat(cursor.next(), is("data")); + assertThat(cursor.getCursorId(), is(2L)); + assertThat(cursor.hasNext(), is(true)); + + assertThat(cursor.next(), is("redis")); + assertThat(cursor.getCursorId(), is(0L)); + assertThat(cursor.hasNext(), is(false)); + } + + /** + * @see DATAREDIS-290 + */ + @Test + public void positionShouldBeIncrementedCorrectly() throws IOException { + + LinkedList> values = new LinkedList>(); + values.add(createIteration(1, "spring")); + values.add(createIteration(2, "data")); + values.add(createIteration(0, "redis")); + Cursor cursor = initCursor(values).open(); + + cursor.open(); + assertThat(cursor.getPosition(), is(0L)); + + cursor.next(); + assertThat(cursor.getPosition(), is(1L)); + + cursor.next(); + assertThat(cursor.getPosition(), is(2L)); + } + + private CapturingCursorDummy initCursor(Queue> values) { + CapturingCursorDummy cursor = new CapturingCursorDummy(values); + cursor.open(); + return cursor; + } + + private ScanIteration createIteration(long cursorId, String... values) { + return new ScanIteration(cursorId, Arrays.asList(values)); + } + + private class CapturingCursorDummy extends ScanCursor { + + private Queue> values; + + private Stack cursors; + + public CapturingCursorDummy(Queue> values) { + this.values = values; + } + + @Override + protected ScanIteration doScan(long cursorId, ScanOptions options) { + + if (cursors == null) { + cursors = new Stack(); + } + this.cursors.push(cursorId); + return this.values.poll(); + } + } +}