DATAREDIS-314 - Add SCAN commands to Collection types.
Add scan to RedisMap / RedisSet / RedisZSet. Lua scritps have to be assembled in binary way in order to avoid script errors when executing. Add missing rules/profiles to run properly in a 2.6.x redis environment. Original pull request: #82.
This commit is contained in:
committed by
Thomas Darimont
parent
7b3a358752
commit
42ff3ee1e1
@@ -17,6 +17,9 @@ package org.springframework.data.redis.connection.lettuce;
|
||||
|
||||
import static com.lambdaworks.redis.protocol.CommandType.*;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
@@ -3105,19 +3108,51 @@ public class LettuceConnection implements RedisConnection {
|
||||
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);
|
||||
byte[] script = createBinaryLuaScriptForScan("HSCAN", key, cursorId, options);
|
||||
List<?> result = eval(script, ReturnType.MULTI, 0);
|
||||
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
|
||||
|
||||
Map<byte[], byte[]> values = failsafeReadScanValues(result, LettuceConverters.bytesListToMapConverter());
|
||||
return new ScanIteration<Entry<byte[], byte[]>>(Long.valueOf(nextCursorId), values.entrySet());
|
||||
}
|
||||
|
||||
}.open();
|
||||
}
|
||||
|
||||
private byte[] createBinaryLuaScriptForScan(String command, byte[] key, long cursorId, ScanOptions options) {
|
||||
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
|
||||
|
||||
try {
|
||||
|
||||
outputStream.write(("return redis.call('" + command + "'").getBytes("UTF-8"));
|
||||
outputStream.write(", '".getBytes("UTF-8"));
|
||||
writeFilteredKey(key, outputStream);
|
||||
outputStream.write("', ".getBytes("UTF-8"));
|
||||
outputStream.write(Long.toString(cursorId).getBytes("UTF-8"));
|
||||
outputStream.write(options.toOptionString().getBytes("UTF-8"));
|
||||
outputStream.write(")".getBytes("UTF-8"));
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
return outputStream.toByteArray();
|
||||
}
|
||||
|
||||
private void writeFilteredKey(byte[] key, OutputStream stream) {
|
||||
byte toBeFiltered = (byte) '\r';
|
||||
for (byte b : key) {
|
||||
try {
|
||||
if (toBeFiltered == b) {
|
||||
stream.write("\\r".getBytes());
|
||||
} else {
|
||||
stream.write(b);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
throw new IllegalArgumentException("Cannot convert key to suitable redis key for lua", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
|
||||
@@ -3145,11 +3180,8 @@ public class LettuceConnection implements RedisConnection {
|
||||
throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
|
||||
}
|
||||
|
||||
String params = " ,'" + LettuceConverters.bytesToString().convert(key) + "', " + cursorId
|
||||
+ options.toOptionString();
|
||||
String script = "return redis.call('SSCAN'" + params + ")";
|
||||
|
||||
List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
|
||||
byte[] script = createBinaryLuaScriptForScan("SSCAN", key, cursorId, options);
|
||||
List<?> result = eval(script, ReturnType.MULTI, 0);
|
||||
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
|
||||
|
||||
List<byte[]> values = failsafeReadScanValues(result, null);
|
||||
@@ -3185,11 +3217,9 @@ public class LettuceConnection implements RedisConnection {
|
||||
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('ZSCAN'" + params + ")";
|
||||
byte[] script = createBinaryLuaScriptForScan("ZSCAN", key, cursorId, options);
|
||||
List<?> result = eval(script, ReturnType.MULTI, 0);
|
||||
|
||||
List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
|
||||
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
|
||||
|
||||
List<Tuple> values = failsafeReadScanValues(result, LettuceConverters.bytesListToTupleListConverter());
|
||||
|
||||
@@ -26,7 +26,9 @@ import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.springframework.data.redis.connection.DataType;
|
||||
import org.springframework.data.redis.core.BoundHashOperations;
|
||||
import org.springframework.data.redis.core.Cursor;
|
||||
import org.springframework.data.redis.core.RedisOperations;
|
||||
import org.springframework.data.redis.core.ScanOptions;
|
||||
import org.springframework.data.redis.core.SessionCallback;
|
||||
|
||||
/**
|
||||
@@ -293,4 +295,22 @@ public class DefaultRedisMap<K, V> implements RedisMap<K, V> {
|
||||
throw new IllegalStateException("Cannot read collection with Redis connection in pipeline/multi-exec mode");
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.support.collections.RedisMap#scan()
|
||||
*/
|
||||
@Override
|
||||
public Cursor<java.util.Map.Entry<K, V>> scan() {
|
||||
return scan(ScanOptions.NONE);
|
||||
}
|
||||
|
||||
/**
|
||||
* @since 1.4
|
||||
* @param options
|
||||
* @return
|
||||
*/
|
||||
private Cursor<java.util.Map.Entry<K, V>> scan(ScanOptions options) {
|
||||
return hashOps.scan(options);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
@@ -23,13 +23,16 @@ import java.util.UUID;
|
||||
|
||||
import org.springframework.data.redis.connection.DataType;
|
||||
import org.springframework.data.redis.core.BoundSetOperations;
|
||||
import org.springframework.data.redis.core.Cursor;
|
||||
import org.springframework.data.redis.core.RedisOperations;
|
||||
import org.springframework.data.redis.core.ScanOptions;
|
||||
|
||||
/**
|
||||
* Default implementation for {@link RedisSet}. Note that the collection support works only with normal,
|
||||
* non-pipeline/multi-exec connections as it requires a reply to be sent right away.
|
||||
*
|
||||
* @author Costin Leau
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public class DefaultRedisSet<E> extends AbstractRedisCollection<E> implements RedisSet<E> {
|
||||
|
||||
@@ -162,4 +165,22 @@ public class DefaultRedisSet<E> extends AbstractRedisCollection<E> implements Re
|
||||
public DataType getType() {
|
||||
return DataType.SET;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.support.collections.RedisSet#scan()
|
||||
*/
|
||||
@Override
|
||||
public Cursor<E> scan() {
|
||||
return scan(ScanOptions.NONE);
|
||||
}
|
||||
|
||||
/**
|
||||
* @since 1.4
|
||||
* @param options
|
||||
* @return
|
||||
*/
|
||||
public Cursor<E> scan(ScanOptions options) {
|
||||
return boundSetOps.scan(options);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
@@ -20,9 +20,13 @@ import java.util.Iterator;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.core.convert.converter.Converter;
|
||||
import org.springframework.data.redis.connection.DataType;
|
||||
import org.springframework.data.redis.core.BoundZSetOperations;
|
||||
import org.springframework.data.redis.core.ConvertingCursor;
|
||||
import org.springframework.data.redis.core.Cursor;
|
||||
import org.springframework.data.redis.core.RedisOperations;
|
||||
import org.springframework.data.redis.core.ScanOptions;
|
||||
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
|
||||
|
||||
/**
|
||||
@@ -30,6 +34,7 @@ import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
|
||||
* non-pipeline/multi-exec connections as it requires a reply to be sent right away.
|
||||
*
|
||||
* @author Costin Leau
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public class DefaultRedisZSet<E> extends AbstractRedisCollection<E> implements RedisZSet<E> {
|
||||
|
||||
@@ -229,4 +234,29 @@ public class DefaultRedisZSet<E> extends AbstractRedisCollection<E> implements R
|
||||
public DataType getType() {
|
||||
return DataType.ZSET;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.support.collections.RedisZSet#scan()
|
||||
*/
|
||||
@Override
|
||||
public Cursor<E> scan() {
|
||||
return new ConvertingCursor<TypedTuple<E>, E>(scan(ScanOptions.NONE), new Converter<TypedTuple<E>, E>() {
|
||||
|
||||
@Override
|
||||
public E convert(TypedTuple<E> source) {
|
||||
return source.getValue();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* @since 1.4
|
||||
* @param options
|
||||
* @return
|
||||
*/
|
||||
public Cursor<TypedTuple<E>> scan(ScanOptions options) {
|
||||
return boundZSetOps.scan(options);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
@@ -15,16 +15,25 @@
|
||||
*/
|
||||
package org.springframework.data.redis.support.collections;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
/**
|
||||
* Map view of a Redis hash.
|
||||
*
|
||||
* @author Costin Leau
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public interface RedisMap<K, V> extends RedisStore, ConcurrentMap<K, V> {
|
||||
|
||||
Long increment(K key, long delta);
|
||||
|
||||
Double increment(K key, double delta);
|
||||
|
||||
/**
|
||||
* @since 1.4
|
||||
* @return
|
||||
*/
|
||||
Iterator<Map.Entry<K, V>> scan();
|
||||
}
|
||||
|
||||
@@ -21,8 +21,10 @@ import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.Date;
|
||||
import java.util.Enumeration;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Properties;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@@ -247,4 +249,9 @@ public class RedisProperties extends Properties implements RedisMap<Object, Obje
|
||||
public synchronized void storeToXML(OutputStream os, String comment) throws IOException {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Iterator<java.util.Map.Entry<Object, Object>> scan() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
@@ -16,12 +16,14 @@
|
||||
package org.springframework.data.redis.support.collections;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* Redis extension for the {@link Set} contract. Supports {@link Set} specific operations backed by Redis operations.
|
||||
*
|
||||
* @author Costin Leau
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public interface RedisSet<E> extends RedisCollection<E>, Set<E> {
|
||||
|
||||
@@ -48,4 +50,10 @@ public interface RedisSet<E> extends RedisCollection<E>, Set<E> {
|
||||
RedisSet<E> diffAndStore(RedisSet<?> set, String destKey);
|
||||
|
||||
RedisSet<E> diffAndStore(Collection<? extends RedisSet<?>> sets, String destKey);
|
||||
|
||||
/**
|
||||
* @since 1.4
|
||||
* @return
|
||||
*/
|
||||
Iterator<E> scan();
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ package org.springframework.data.redis.support.collections;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Comparator;
|
||||
import java.util.Iterator;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.Set;
|
||||
import java.util.SortedSet;
|
||||
@@ -124,4 +125,10 @@ public interface RedisZSet<E> extends RedisCollection<E>, Set<E> {
|
||||
* @throws NoSuchElementException sorted set is empty.
|
||||
*/
|
||||
E last();
|
||||
|
||||
/**
|
||||
* @since 1.4
|
||||
* @return
|
||||
*/
|
||||
Iterator<E> scan();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user