Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-306 - Add support for ZSCAN.

Browse Source 
ZSCAN is directly supported by jedis and can be emulated for lettuce via eval.

SRP and JRedis will throw UnsupportedOperationException.

Original pull request: #80.
This commit is contained in:
Christoph Strobl
2014-06-02 21:35:05 +02:00
committed by Thomas Darimont
parent f6e88989f1
commit c939c8a9ff
16 changed files with 346 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
View File

@@ -2228,6 +2228,14 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
return this.delegate.scan(options);
}
/*
*
*/
@Override
public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
return this.delegate.zScan(key, options);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSetCommands#scan(byte[], org.springframework.data.redis.core.ScanOptions)
@@ -2320,24 +2328,6 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
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);
}
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.StringRedisConnection#hScan(java.lang.String, org.springframework.data.redis.core.ScanOptions)
@@ -2354,12 +2344,12 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
@Override
public String getKey() {
return DefaultStringRedisConnection.this.serializer.deserialize(source.getKey());
return bytesToString.convert(source.getKey());
}
@Override
public String getValue() {
return DefaultStringRedisConnection.this.serializer.deserialize(source.getValue());
return bytesToString.convert(source.getValue());
}
@Override
@@ -2370,4 +2360,24 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
}
});
}
/*
* (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), bytesToString);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.StringRedisConnection#zScan(java.lang.String, org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<StringTuple> zScan(String key, ScanOptions options) {
return new ConvertingCursor<Tuple, StringRedisConnection.StringTuple>(delegate.zScan(this.serialize(key), options),
new TupleConverter());
}
}

13
src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java
View File

@@ -17,6 +17,9 @@ package org.springframework.data.redis.connection;
import java.util.Set;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
/**
* ZSet(SortedSet)-specific commands supported by Redis.
*
@@ -332,4 +335,14 @@ public interface RedisZSetCommands {
* @return
*/
Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets);
/**
* Use a {@link Cursor} to iterate over elements in sorted set at {@code key}.
*
* @param key
* @param options
* @return
* @since 1.4
*/
Cursor<Tuple> zScan(byte[] key, ScanOptions options);
}

12
src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java
View File

@@ -319,6 +319,15 @@ public interface StringRedisConnection extends RedisConnection {
*/
List<RedisClientInfo> getClientList();
/**
* @see RedisHashCommands#hScan(byte[], ScanOptions)
* @param key
* @param options
* @return
* @since 1.4
*/
Cursor<Map.Entry<String, String>> hScan(String key, ScanOptions options);
/**
* @see RedisSetCommands#sScan(byte[], ScanOptions)
* @param key
@@ -329,10 +338,11 @@ public interface StringRedisConnection extends RedisConnection {
Cursor<String> sScan(String key, ScanOptions options);
/**
* @see RedisZSetCommands#zScan(byte[], ScanOptions)
* @param key
* @param options
* @return
* @since 1.4
*/
Cursor<Map.Entry<String, String>> hScan(String key, ScanOptions options);
Cursor<StringTuple> zScan(String key, ScanOptions options);
}

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

@@ -38,6 +38,7 @@ import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisPipelineException;
import org.springframework.data.redis.connection.RedisSubscribedConnectionException;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.SortParameters;
import org.springframework.data.redis.connection.Subscription;
@@ -2949,6 +2950,43 @@ public class JedisConnection implements RedisConnection {
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisZSetCommands#zScan(byte[], org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
return zScan(key, 0L, options);
}
/**
* @param key
* @param cursorId
* @param options
* @return
* @since 1.4
*/
public Cursor<Tuple> zScan(byte[] key, Long cursorId, ScanOptions options) {
return new KeyBoundCursor<Tuple>(key, cursorId, options) {
@Override
protected ScanIteration<Tuple> doScan(byte[] key, long cursorId, ScanOptions options) {
if (isQueueing() || isPipelined()) {
throw new UnsupportedOperationException("'ZSCAN' cannot be called in pipeline / transaction mode.");
}
ScanParams params = prepareScanParams(options);
ScanResult<redis.clients.jedis.Tuple> result = jedis.zscan(key, JedisConverters.toBytes(cursorId), params);
return new ScanIteration<RedisZSetCommands.Tuple>(Long.valueOf(result.getStringCursor()), JedisConverters
.tuplesToTuples().convert(result.getResult()));
}
}.open();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)

25
src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java
View File

@@ -59,6 +59,8 @@ abstract public class JedisConverters extends Converters {
private static final SetConverter<redis.clients.jedis.Tuple, Tuple> TUPLE_SET_TO_TUPLE_SET;
private static final Converter<Exception, DataAccessException> EXCEPTION_CONVERTER = new JedisExceptionConverter();
private static final Converter<String[], List<RedisClientInfo>> STRING_TO_CLIENT_INFO_CONVERTER = new StringToRedisClientInfoConverter();
private static final Converter<redis.clients.jedis.Tuple, Tuple> TUPLE_CONVERTER;
private static final ListConverter<redis.clients.jedis.Tuple, Tuple> TUPLE_LIST_TO_TUPLE_LIST_CONVERTER;
static {
STRING_TO_BYTES = new Converter<String, byte[]>() {
@@ -69,19 +71,30 @@ abstract public class JedisConverters extends Converters {
STRING_LIST_TO_BYTE_LIST = new ListConverter<String, byte[]>(STRING_TO_BYTES);
STRING_SET_TO_BYTE_SET = new SetConverter<String, byte[]>(STRING_TO_BYTES);
STRING_MAP_TO_BYTE_MAP = new MapConverter<String, byte[]>(STRING_TO_BYTES);
TUPLE_SET_TO_TUPLE_SET = new SetConverter<redis.clients.jedis.Tuple, Tuple>(
new Converter<redis.clients.jedis.Tuple, Tuple>() {
public Tuple convert(redis.clients.jedis.Tuple source) {
return source != null ? new DefaultTuple(source.getBinaryElement(), source.getScore()) : null;
}
TUPLE_CONVERTER = new Converter<redis.clients.jedis.Tuple, Tuple>() {
public Tuple convert(redis.clients.jedis.Tuple source) {
return source != null ? new DefaultTuple(source.getBinaryElement(), source.getScore()) : null;
}
});
};
TUPLE_SET_TO_TUPLE_SET = new SetConverter<redis.clients.jedis.Tuple, Tuple>(TUPLE_CONVERTER);
TUPLE_LIST_TO_TUPLE_LIST_CONVERTER = new ListConverter<redis.clients.jedis.Tuple, Tuple>(TUPLE_CONVERTER);
}
public static Converter<String, byte[]> stringToBytes() {
return STRING_TO_BYTES;
}
/**
* {@link ListConverter} converting jedis {@link redis.clients.jedis.Tuple} to {@link Tuple}.
*
* @return
* @since 1.4
*/
public static ListConverter<redis.clients.jedis.Tuple, Tuple> tuplesToTuples() {
return TUPLE_LIST_TO_TUPLE_LIST_CONVERTER;
}
public static ListConverter<String, byte[]> stringListToByteList() {
return STRING_LIST_TO_BYTE_LIST;
}

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

@@ -1249,6 +1249,15 @@ public class JredisConnection implements RedisConnection {
throw new UnsupportedOperationException("'SCAN' command is not supported for jredis.");
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisZSetCommands#zScan(byte[], org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
throw new UnsupportedOperationException("'ZSCAN' command is not supported for jredis.");
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)

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

@@ -3069,12 +3069,52 @@ public class LettuceConnection implements RedisConnection {
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)));
return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), ((List<byte[]>) result.get(1)));
}
}.open();
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisHashCommands#hScan(byte[], org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, ScanOptions options) {
return hscan(key, 0, options);
}
/**
* @param key
* @param cursorId
* @param options
* @return
* @since 1.4
*/
public Cursor<Entry<byte[], byte[]>> hscan(byte[] key, long cursorId, ScanOptions options) {
return new KeyBoundCursor<Entry<byte[], byte[]>>(key, cursorId, options) {
@Override
protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, long cursorId, ScanOptions options) {
if (isQueueing() || isPipelined()) {
throw new UnsupportedOperationException("'HSCAN' cannot be called in pipeline / transaction mode.");
}
String params = " ,'" + LettuceConverters.bytesToString().convert(key) + "', " + cursorId
+ options.toOptionString();
String script = "return redis.call('HSCAN'" + params + ")";
List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
@SuppressWarnings("unchecked")
Map<byte[], byte[]> values = LettuceConverters.toMap((List<byte[]>) result.get(1));
return new ScanIteration<Entry<byte[], byte[]>>(Long.valueOf(nextCursorId), values.entrySet());
}
}.open();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
@@ -3115,12 +3155,13 @@ public class LettuceConnection implements RedisConnection {
}.open();
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisHashCommands#hScan(byte[], org.springframework.data.redis.core.ScanOptions)
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisZSetCommands#zScan(byte[], org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, ScanOptions options) {
return hscan(key, 0, options);
public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
return zScan(key, 0L, options);
}
/**
@@ -3130,27 +3171,27 @@ public class LettuceConnection implements RedisConnection {
* @return
* @since 1.4
*/
public Cursor<Entry<byte[], byte[]>> hscan(byte[] key, long cursorId, ScanOptions options) {
public Cursor<Tuple> zScan(byte[] key, long cursorId, ScanOptions options) {
return new KeyBoundCursor<Entry<byte[], byte[]>>(key, cursorId, options) {
return new KeyBoundCursor<Tuple>(key, cursorId, options) {
@SuppressWarnings("unchecked")
@Override
protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, long cursorId, ScanOptions options) {
protected ScanIteration<Tuple> doScan(byte[] key, long cursorId, ScanOptions options) {
if (isQueueing() || isPipelined()) {
throw new UnsupportedOperationException("'HSCAN' cannot be called in pipeline / transaction mode.");
throw new UnsupportedOperationException("'ZSCAN' cannot be called in pipeline / transaction mode.");
}
String params = " ,'" + LettuceConverters.bytesToString().convert(key) + "', " + cursorId
+ options.toOptionString();
String script = "return redis.call('HSCAN'" + params + ")";
String script = "return redis.call('ZSCAN'" + params + ")";
List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
@SuppressWarnings("unchecked")
Map<byte[], byte[]> values = LettuceConverters.toMap((List<byte[]>) result.get(1));
return new ScanIteration<Entry<byte[], byte[]>>(Long.valueOf(nextCursorId), values.entrySet());
return new ScanIteration<Tuple>(Long.valueOf(nextCursorId), LettuceConverters.toTuple((List<byte[]>) result
.get(1)));
}
}.open();
}

24
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java
View File

@@ -67,6 +67,7 @@ abstract public class LettuceConverters extends Converters {
private static final Converter<Exception, DataAccessException> EXCEPTION_CONVERTER = new LettuceExceptionConverter();
private static final Converter<Long, Boolean> LONG_TO_BOOLEAN = new LongToBooleanConverter();
private static final Converter<List<byte[]>, Map<byte[], byte[]>> BYTES_LIST_TO_MAP;
private static final Converter<List<byte[]>, List<Tuple>> BYTES_LIST_TO_TUPLE_LIST_CONVERTER;
private static final Converter<String[], List<RedisClientInfo>> STRING_TO_LIST_OF_CLIENT_INFO = new StringToRedisClientInfoConverter();
@@ -148,7 +149,29 @@ abstract public class LettuceConverters extends Converters {
}
};
BYTES_LIST_TO_TUPLE_LIST_CONVERTER = new Converter<List<byte[]>, List<Tuple>>() {
@Override
public List<Tuple> convert(List<byte[]> source) {
if (CollectionUtils.isEmpty(source)) {
return Collections.emptyList();
}
List<Tuple> tuples = new ArrayList<Tuple>();
Iterator<byte[]> it = source.iterator();
while (it.hasNext()) {
tuples.add(new DefaultTuple(it.next(), it.hasNext() ? Double.valueOf(LettuceConverters.toString(it.next()))
: null));
}
return tuples;
}
};
}
public static List<Tuple> toTuple(List<byte[]> list) {
return BYTES_LIST_TO_TUPLE_LIST_CONVERTER.convert(list);
}
public static Converter<String, List<RedisClientInfo>> stringToRedisClientListConverter() {
@@ -301,4 +324,5 @@ abstract public class LettuceConverters extends Converters {
public static List<RedisClientInfo> toListOfRedisClientInformation(String clientList) {
return stringToRedisClientListConverter().convert(clientList);
}
}

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

@@ -2343,6 +2343,11 @@ public class SrpConnection implements RedisConnection {
throw new UnsupportedOperationException("'SCAN' command is not supported for Srp.");
}
@Override
public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
throw new UnsupportedOperationException("'ZSCAN' command is not supported for Srp.");
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)

21
src/main/java/org/springframework/data/redis/core/AbstractOperations.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2011-2013 the original author or authors.
* Copyright 2011-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.
@@ -35,6 +35,7 @@ import org.springframework.util.Assert;
*
* @author Costin Leau
* @author Jennifer Hickey
* @author Christoph Strobl
*/
abstract class AbstractOperations<K, V> {
@@ -181,22 +182,26 @@ abstract class AbstractOperations<K, V> {
return SerializationUtils.deserialize(rawValues, valueSerializer());
}
@SuppressWarnings({ "unchecked", "rawtypes" })
Set<TypedTuple<V>> deserializeTupleValues(Set<Tuple> rawValues) {
Set<TypedTuple<V>> deserializeTupleValues(Collection<Tuple> rawValues) {
if (rawValues == null) {
return null;
}
Set<TypedTuple<V>> set = new LinkedHashSet<TypedTuple<V>>(rawValues.size());
for (Tuple rawValue : rawValues) {
Object value = rawValue.getValue();
if (valueSerializer() != null) {
value = valueSerializer().deserialize(rawValue.getValue());
}
set.add(new DefaultTypedTuple(value, rawValue.getScore()));
set.add(deserializeTuple(rawValue));
}
return set;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
TypedTuple<V> deserializeTuple(Tuple tuple) {
Object value = tuple.getValue();
if (valueSerializer() != null) {
value = valueSerializer().deserialize(tuple.getValue());
}
return new DefaultTypedTuple(value, tuple.getScore());
}
@SuppressWarnings("unchecked")
Set<Tuple> rawTupleValues(Set<TypedTuple<V>> values) {
if (values == null) {

8
src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java
View File

@@ -17,6 +17,7 @@
package org.springframework.data.redis.core;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
@@ -90,4 +91,11 @@ public interface BoundZSetOperations<K, V> extends BoundKeyOperations<K> {
Long zCard();
Double score(Object o);
/**
* @param options
* @return
* @since 1.4
*/
Iterator<TypedTuple<V>> scan(ScanOptions options);
}

10
src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java
View File

@@ -17,6 +17,7 @@
package org.springframework.data.redis.core;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import org.springframework.data.redis.connection.DataType;
@@ -156,4 +157,13 @@ class DefaultBoundZSetOperations<K, V> extends DefaultBoundKeyOperations<K> impl
public DataType getType() {
return DataType.ZSET;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.core.BoundZSetOperations#scan(org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Iterator<TypedTuple<V>> scan(ScanOptions options) {
return ops.scan(getKey(), options);
}
}

28
src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java
View File

@@ -17,8 +17,11 @@ package org.springframework.data.redis.core;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.Set;
import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
@@ -365,4 +368,29 @@ class DefaultZSetOperations<K, V> extends AbstractOperations<K, V> implements ZS
}
}, true);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.core.ZSetOperations#scan(java.lang.Object, org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Iterator<TypedTuple<V>> scan(K key, final ScanOptions options) {
final byte[] rawKey = rawKey(key);
Cursor<Tuple> cursor = execute(new RedisCallback<Cursor<Tuple>>() {
@Override
public Cursor<Tuple> doInRedis(RedisConnection connection) throws DataAccessException {
return connection.zScan(rawKey, options);
}
}, true);
return new ConvertingCursor<Tuple, TypedTuple<V>>(cursor, new Converter<Tuple, TypedTuple<V>>() {
@Override
public TypedTuple<V> convert(Tuple source) {
return deserializeTuple(source);
}
});
}
}

9
src/main/java/org/springframework/data/redis/core/ZSetOperations.java
View File

@@ -17,6 +17,7 @@
package org.springframework.data.redis.core;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
/**
@@ -107,4 +108,12 @@ public interface ZSetOperations<K, V> {
Long zCard(K key);
RedisOperations<K, V> getOperations();
/**
* @param key
* @param options
* @return
* @since 1.4
*/
Iterator<TypedTuple<V>> scan(K key, ScanOptions options);
}