rik/jlibredis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merging with upstream, necessary changes, shifting poolConfig as private instance variable from JedisClusterConnectionHandler to JedisClusterInfoCache due to design change in previous commits.

Browse Source
This commit is contained in:
Mayank Dang
2014-09-11 16:16:08 +05:30
parent 19fc1f87cf fc594db9a3
commit 37a3c7b835
55 changed files with 1506 additions and 727 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
src/main/java/redis/clients/jedis/BinaryClient.java
View File

@@ -7,6 +7,7 @@ import redis.clients.util.SafeEncoder;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import static redis.clients.jedis.Protocol.Command.*;
@@ -827,6 +828,25 @@ public class BinaryClient extends Connection {
args.addAll(params.getParams());
sendCommand(ZINTERSTORE, args.toArray(new byte[args.size()][]));
}
public void zlexcount(final byte[] key, final byte[] min, final byte[] max) {
sendCommand(ZLEXCOUNT, key, min, max);
}
public void zrangeByLex(final byte[] key, final byte[] min, final byte[] max) {
sendCommand(ZRANGEBYLEX, key, min, max);
}
public void zrangeByLex(final byte[] key, final byte[] min, final byte[] max,
final int offset, final int count) {
sendCommand(ZRANGEBYLEX, key, min, max, LIMIT.raw,
toByteArray(offset), toByteArray(count));
}
public void zremrangeByLex(byte[] key, byte[] min, byte[] max) {
sendCommand(ZREMRANGEBYLEX, key, min, max);
}
public void save() {
sendCommand(SAVE);
@@ -963,9 +983,6 @@ public class BinaryClient extends Connection {
}
public void resetState() {
if (isInMulti())
discard();
if (isInWatch())
unwatch();
}
@@ -1099,11 +1116,6 @@ public class BinaryClient extends Connection {
sendCommand(RESTORE, key, toByteArray(ttl), serializedValue);
}
@Deprecated
public void pexpire(final byte[] key, final int milliseconds) {
pexpire(key, (long) milliseconds);
}
public void pexpire(final byte[] key, final long milliseconds) {
sendCommand(PEXPIRE, key, toByteArray(milliseconds));
}
@@ -1165,61 +1177,6 @@ public class BinaryClient extends Connection {
sendCommand(HINCRBYFLOAT, key, field, toByteArray(increment));
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void scan(int cursor, final ScanParams params) {
final List<byte[]> args = new ArrayList<byte[]>();
args.add(toByteArray(cursor));
args.addAll(params.getParams());
sendCommand(SCAN, args.toArray(new byte[args.size()][]));
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void hscan(final byte[] key, int cursor, final ScanParams params) {
final List<byte[]> args = new ArrayList<byte[]>();
args.add(key);
args.add(toByteArray(cursor));
args.addAll(params.getParams());
sendCommand(HSCAN, args.toArray(new byte[args.size()][]));
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void sscan(final byte[] key, int cursor, final ScanParams params) {
final List<byte[]> args = new ArrayList<byte[]>();
args.add(key);
args.add(toByteArray(cursor));
args.addAll(params.getParams());
sendCommand(SSCAN, args.toArray(new byte[args.size()][]));
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void zscan(final byte[] key, int cursor, final ScanParams params) {
final List<byte[]> args = new ArrayList<byte[]>();
args.add(key);
args.add(toByteArray(cursor));
args.addAll(params.getParams());
sendCommand(ZSCAN, args.toArray(new byte[args.size()][]));
}
public void scan(final byte[] cursor, final ScanParams params) {
final List<byte[]> args = new ArrayList<byte[]>();
args.add(cursor);

93
src/main/java/redis/clients/jedis/BinaryJedis.java
View File

@@ -17,21 +17,20 @@ import redis.clients.jedis.BinaryClient.LIST_POSITION;
import redis.clients.jedis.exceptions.JedisDataException;
import redis.clients.jedis.exceptions.JedisException;
import redis.clients.util.JedisByteHashMap;
import redis.clients.util.JedisURIHelper;
import redis.clients.util.SafeEncoder;
public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
MultiKeyBinaryCommands, AdvancedBinaryJedisCommands,
BinaryScriptingCommands, Closeable {
protected Client client = null;
protected Transaction transaction = null;
protected Pipeline pipeline = null;
public BinaryJedis(final String host) {
URI uri = URI.create(host);
if (uri.getScheme() != null && uri.getScheme().equals("redis")) {
client = new Client(uri.getHost(), uri.getPort());
client.auth(uri.getUserInfo().split(":", 2)[1]);
client.getStatusCodeReply();
client.select(Integer.parseInt(uri.getPath().split("/", 2)[1]));
client.getStatusCodeReply();
initializeClientFromURI(uri);
} else {
client = new Client(host);
}
@@ -53,11 +52,28 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
}
public BinaryJedis(URI uri) {
initializeClientFromURI(uri);
}
public BinaryJedis(final URI uri, final int timeout) {
initializeClientFromURI(uri);
client.setTimeout(timeout);
}
private void initializeClientFromURI(URI uri) {
client = new Client(uri.getHost(), uri.getPort());
client.auth(uri.getUserInfo().split(":", 2)[1]);
client.getStatusCodeReply();
client.select(Integer.parseInt(uri.getPath().split("/", 2)[1]));
client.getStatusCodeReply();
String password = JedisURIHelper.getPassword(uri);
if (password != null) {
client.auth(password);
client.getStatusCodeReply();
}
Integer dbIndex = JedisURIHelper.getDBIndex(uri);
if (dbIndex > 0) {
client.select(dbIndex);
client.getStatusCodeReply();
}
}
public String ping() {
@@ -1753,7 +1769,9 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
public Transaction multi() {
client.multi();
return new Transaction(client);
client.getOne(); // expected OK
transaction = new Transaction(client);
return transaction;
}
@Deprecated
@@ -1767,6 +1785,7 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
List<Object> results = null;
jedisTransaction.setClient(client);
client.multi();
client.getOne(); // expected OK
jedisTransaction.execute();
results = jedisTransaction.exec();
return results;
@@ -1789,9 +1808,23 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
public void resetState() {
if (client.isConnected()) {
if (transaction != null) {
transaction.clear();
}
if (pipeline != null) {
pipeline.clear();
}
if (client.isInWatch()) {
unwatch();
}
client.resetState();
client.getAll();
}
transaction = null;
pipeline = null;
}
public String watch(final byte[]... keys) {
@@ -2207,7 +2240,7 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
}
public Pipeline pipelined() {
Pipeline pipeline = new Pipeline();
pipeline = new Pipeline();
pipeline.setClient(client);
return pipeline;
}
@@ -2793,6 +2826,35 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
return client.getIntegerReply();
}
@Override
public Long zlexcount(final byte[] key, final byte[] min, final byte[] max) {
checkIsInMulti();
client.zlexcount(key, min, max);
return client.getIntegerReply();
}
@Override
public Set<byte[]> zrangeByLex(final byte[] key, final byte[] min, final byte[] max) {
checkIsInMulti();
client.zrangeByLex(key, min, max);
return new LinkedHashSet<byte[]>(client.getBinaryMultiBulkReply());
}
@Override
public Set<byte[]> zrangeByLex(final byte[] key, final byte[] min, final byte[] max,
final int offset, final int count) {
checkIsInMulti();
client.zrangeByLex(key, min, max, offset, count);
return new LinkedHashSet<byte[]>(client.getBinaryMultiBulkReply());
}
@Override
public Long zremrangeByLex(final byte[] key, final byte[] min, final byte[] max) {
checkIsInMulti();
client.zremrangeByLex(key, min, max);
return client.getIntegerReply();
}
/**
* Synchronously save the DB on disk.
* <p>
@@ -3385,11 +3447,6 @@ public class BinaryJedis implements BasicCommands, BinaryJedisCommands,
return client.getStatusCodeReply();
}
@Deprecated
public Long pexpire(final byte[] key, final int milliseconds) {
return pexpire(key, (long) milliseconds);
}
public Long pexpire(final byte[] key, final long milliseconds) {
checkIsInMulti();
client.pexpire(key, milliseconds);

11
src/main/java/redis/clients/jedis/BinaryJedisCommands.java
View File

@@ -115,6 +115,8 @@ public interface BinaryJedisCommands {
byte[] srandmember(byte[] key);
List<byte[]> srandmember(final byte[] key, final int count);
Long strlen(byte[] key);
Long zadd(byte[] key, double score, byte[] member);
@@ -194,6 +196,15 @@ public interface BinaryJedisCommands {
Long zremrangeByScore(byte[] key, double start, double end);
Long zremrangeByScore(byte[] key, byte[] start, byte[] end);
Long zlexcount(final byte[] key, final byte[] min, final byte[] max);
Set<byte[]> zrangeByLex(final byte[] key, final byte[] min, final byte[] max);
Set<byte[]> zrangeByLex(final byte[] key, final byte[] min, final byte[] max,
int offset, int count);
Long zremrangeByLex(final byte[] key, final byte[] min, final byte[] max);
Long linsert(byte[] key, Client.LIST_POSITION where, byte[] pivot,
byte[] value);

11
src/main/java/redis/clients/jedis/BinaryRedisPipeline.java
View File

@@ -34,7 +34,7 @@ public interface BinaryRedisPipeline {
Response<byte[]> getSet(byte[] key, byte[] value);
Response<Long> getrange(byte[] key, long startOffset, long endOffset);
Response<byte[]> getrange(byte[] key, long startOffset, long endOffset);
Response<Long> hdel(byte[] key, byte[]... field);
@@ -206,6 +206,15 @@ public interface BinaryRedisPipeline {
Response<Long> zrevrank(byte[] key, byte[] member);
Response<Double> zscore(byte[] key, byte[] member);
Response<Long> zlexcount(final byte[] key, final byte[] min, final byte[] max);
Response<Set<byte[]>> zrangeByLex(final byte[] key, final byte[] max, final byte[] min);
Response<Set<byte[]>> zrangeByLex(final byte[] key, final byte[] max, final byte[] min,
int offset, int count);
Response<Long> zremrangeByLex(final byte[] key, final byte[] min, final byte[] max);
Response<Long> bitcount(byte[] key);

31
src/main/java/redis/clients/jedis/BinaryShardedJedis.java
View File

@@ -300,6 +300,12 @@ public class BinaryShardedJedis extends Sharded<Jedis, JedisShardInfo>
return j.srandmember(key);
}
@Override
public List srandmember(byte[] key, int count) {
Jedis j = getShard(key);
return j.srandmember(key, count);
}
public Long zadd(byte[] key, double score, byte[] member) {
Jedis j = getShard(key);
return j.zadd(key, score, member);
@@ -484,6 +490,31 @@ public class BinaryShardedJedis extends Sharded<Jedis, JedisShardInfo>
Jedis j = getShard(key);
return j.zremrangeByScore(key, start, end);
}
@Override
public Long zlexcount(final byte[] key, final byte[] min, final byte[] max) {
Jedis j = getShard(key);
return j.zlexcount(key, min, max);
}
@Override
public Set<byte[]> zrangeByLex(final byte[] key, final byte[] min, final byte[] max) {
Jedis j = getShard(key);
return j.zrangeByLex(key, min, max);
}
@Override
public Set<byte[]> zrangeByLex(final byte[] key, final byte[] min, final byte[] max,
final int offset, final int count) {
Jedis j = getShard(key);
return j.zrangeByLex(key, min, max, offset, count);
}
@Override
public Long zremrangeByLex(final byte[] key, final byte[] min, final byte[] max) {
Jedis j = getShard(key);
return j.zremrangeByLex(key, min, max);
}
public Long linsert(byte[] key, LIST_POSITION where, byte[] pivot,
byte[] value) {

23
src/main/java/redis/clients/jedis/BuilderFactory.java
View File

@@ -9,6 +9,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import redis.clients.util.JedisByteHashMap;
import redis.clients.util.SafeEncoder;
public class BuilderFactory {
@@ -104,6 +105,26 @@ public class BuilderFactory {
};
public static final Builder<Map<String, String>> PUBSUB_NUMSUB_MAP = new Builder<Map<String, String>>() {
@SuppressWarnings("unchecked")
public Map<String, String> build(Object data) {
final List<Object> flatHash = (List<Object>) data;
final Map<String, String> hash = new HashMap<String, String>();
final Iterator<Object> iterator = flatHash.iterator();
while (iterator.hasNext()) {
hash.put(SafeEncoder.encode((byte[]) iterator.next()),
String.valueOf((Long) iterator.next()));
}
return hash;
}
public String toString() {
return "PUBSUB_NUMSUB_MAP<String, String>";
}
};
public static final Builder<Set<String>> STRING_SET = new Builder<Set<String>>() {
@SuppressWarnings("unchecked")
public Set<String> build(Object data) {
@@ -170,7 +191,7 @@ public class BuilderFactory {
@SuppressWarnings("unchecked")
public Map<byte[], byte[]> build(Object data) {
final List<byte[]> flatHash = (List<byte[]>) data;
final Map<byte[], byte[]> hash = new HashMap<byte[], byte[]>();
final Map<byte[], byte[]> hash = new JedisByteHashMap();
final Iterator<byte[]> iterator = flatHash.iterator();
while (iterator.hasNext()) {
hash.put(iterator.next(), iterator.next());

57
src/main/java/redis/clients/jedis/Client.java
View File

@@ -585,6 +585,24 @@ public class Client extends BinaryClient implements Commands {
}
zinterstore(SafeEncoder.encode(dstkey), params, bsets);
}
public void zlexcount(final String key, final String min, final String max) {
zlexcount(SafeEncoder.encode(key), SafeEncoder.encode(min), SafeEncoder.encode(max));
}
public void zrangeByLex(final String key, final String min, final String max) {
zrangeByLex(SafeEncoder.encode(key), SafeEncoder.encode(min), SafeEncoder.encode(max));
}
public void zrangeByLex(final String key, final String min, final String max,
final int offset, final int count) {
zrangeByLex(SafeEncoder.encode(key), SafeEncoder.encode(min), SafeEncoder.encode(max),
offset, count);
}
public void zremrangeByLex(final String key, final String min, final String max) {
zremrangeByLex(SafeEncoder.encode(key), SafeEncoder.encode(min), SafeEncoder.encode(max));
}
public void strlen(final String key) {
strlen(SafeEncoder.encode(key));
@@ -780,11 +798,6 @@ public class Client extends BinaryClient implements Commands {
restore(SafeEncoder.encode(key), ttl, serializedValue);
}
@Deprecated
public void pexpire(final String key, final int milliseconds) {
pexpire(key, (long) milliseconds);
}
public void pexpire(final String key, final long milliseconds) {
pexpire(SafeEncoder.encode(key), milliseconds);
}
@@ -841,36 +854,6 @@ public class Client extends BinaryClient implements Commands {
increment);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void hscan(final String key, int cursor, final ScanParams params) {
hscan(SafeEncoder.encode(key), cursor, params);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void sscan(final String key, int cursor, final ScanParams params) {
sscan(SafeEncoder.encode(key), cursor, params);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void zscan(final String key, int cursor, final ScanParams params) {
zscan(SafeEncoder.encode(key), cursor, params);
}
public void scan(final String cursor, final ScanParams params) {
scan(SafeEncoder.encode(cursor), params);
}
@@ -1011,4 +994,8 @@ public void clusterSetSlotStable(final int slot) {
public void clusterFailover() {
cluster(Protocol.CLUSTER_FAILOVER);
}
public void clusterSlots() {
cluster(Protocol.CLUSTER_SLOTS);
}
}

2
src/main/java/redis/clients/jedis/ClusterCommands.java
View File

@@ -38,4 +38,6 @@ public interface ClusterCommands {
List<String> clusterSlaves(final String nodeId);
String clusterFailover();
List<Object> clusterSlots();
}

32
src/main/java/redis/clients/jedis/Commands.java
View File

@@ -301,38 +301,6 @@ public interface Commands {
public void bitop(BitOP op, final String destKey, String... srcKeys);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void scan(int cursor, final ScanParams params);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void hscan(final String key, int cursor, final ScanParams params);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void sscan(final String key, int cursor, final ScanParams params);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public void zscan(final String key, int cursor, final ScanParams params);
public void scan(final String cursor, final ScanParams params);
public void hscan(final String key, final String cursor, final ScanParams params);

49
src/main/java/redis/clients/jedis/Connection.java
View File

@@ -21,7 +21,6 @@ public class Connection implements Closeable {
private Socket socket;
private RedisOutputStream outputStream;
private RedisInputStream inputStream;
private int pipelinedCommands = 0;
private int timeout = Protocol.DEFAULT_TIMEOUT;
private boolean broken = false;
@@ -78,7 +77,6 @@ public class Connection implements Closeable {
try {
connect();
Protocol.sendCommand(outputStream, cmd, args);
pipelinedCommands++;
return this;
} catch (JedisConnectionException ex) {
// Any other exceptions related to connection?
@@ -91,7 +89,6 @@ public class Connection implements Closeable {
try {
connect();
Protocol.sendCommand(outputStream, cmd, new byte[0][]);
pipelinedCommands++;
return this;
} catch (JedisConnectionException ex) {
// Any other exceptions related to connection?
@@ -180,7 +177,6 @@ public class Connection implements Closeable {
protected String getStatusCodeReply() {
flush();
pipelinedCommands--;
final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
if (null == resp) {
return null;
@@ -200,13 +196,11 @@ public class Connection implements Closeable {
public byte[] getBinaryBulkReply() {
flush();
pipelinedCommands--;
return (byte[]) readProtocolWithCheckingBroken();
}
public Long getIntegerReply() {
flush();
pipelinedCommands--;
return (Long) readProtocolWithCheckingBroken();
}
@@ -217,14 +211,9 @@ public class Connection implements Closeable {
@SuppressWarnings("unchecked")
public List<byte[]> getBinaryMultiBulkReply() {
flush();
pipelinedCommands--;
return (List<byte[]>) readProtocolWithCheckingBroken();
}
public void resetPipelinedCount() {
pipelinedCommands = 0;
}
@SuppressWarnings("unchecked")
public List<Object> getRawObjectMultiBulkReply() {
return (List<Object>) readProtocolWithCheckingBroken();
@@ -232,38 +221,17 @@ public class Connection implements Closeable {
public List<Object> getObjectMultiBulkReply() {
flush();
pipelinedCommands--;
return getRawObjectMultiBulkReply();
}
@SuppressWarnings("unchecked")
public List<Long> getIntegerMultiBulkReply() {
flush();
pipelinedCommands--;
return (List<Long>) readProtocolWithCheckingBroken();
}
public List<Object> getAll() {
return getAll(0);
}
public List<Object> getAll(int except) {
List<Object> all = new ArrayList<Object>();
flush();
while (pipelinedCommands > except) {
try {
all.add(readProtocolWithCheckingBroken());
} catch (JedisDataException e) {
all.add(e);
}
pipelinedCommands--;
}
return all;
flush();
return (List<Long>) Protocol.read(inputStream);
}
public Object getOne() {
flush();
pipelinedCommands--;
return readProtocolWithCheckingBroken();
}
@@ -288,4 +256,17 @@ public class Connection implements Closeable {
throw exc;
}
}
public List<Object> getMany(int count) {
flush();
List<Object> responses = new ArrayList<Object>();
for (int i = 0; i < count; i++) {
try {
responses.add(readProtocolWithCheckingBroken());
} catch (JedisDataException e) {
responses.add(e);
}
}
return responses;
}
}

279
src/main/java/redis/clients/jedis/Jedis.java
View File

@@ -35,6 +35,10 @@ public class Jedis extends BinaryJedis implements JedisCommands,
super(uri);
}
public Jedis(final URI uri, final int timeout) {
super(uri, timeout);
}
/**
* Set the string value as value of the key. The string can't be longer than
* 1073741824 bytes (1 GB).
@@ -547,26 +551,27 @@ public class Jedis extends BinaryJedis implements JedisCommands,
/**
* INCRBYFLOAT
* <p>
* INCRBYFLOAT commands are limited to double precision floating point values.
* INCRBYFLOAT commands are limited to double precision floating point
* values.
* <p>
* Note: this is actually a string operation, that is, in Redis there are
* not "double" types. Simply the string stored at the key is parsed as a
* base double precision floating point value, incremented, and then
* converted back as a string. There is no DECRYBYFLOAT but providing a
* converted back as a string. There is no DECRYBYFLOAT but providing a
* negative value will work as expected.
* <p>
* Time complexity: O(1)
*
*
* @param key
* @param value
* @return Double reply, this commands will reply with the new value of key
* after the increment.
*/
public Double incrByFloat(final String key, final double value) {
checkIsInMulti();
client.incrByFloat(key, value);
String dval = client.getBulkReply();
return (dval != null ? new Double(dval) : null);
checkIsInMulti();
client.incrByFloat(key, value);
String dval = client.getBulkReply();
return (dval != null ? new Double(dval) : null);
}
/**
@@ -763,28 +768,29 @@ public class Jedis extends BinaryJedis implements JedisCommands,
/**
* Increment the number stored at field in the hash at key by a double
* precision floating point value. If key does not exist,
* a new key holding a hash is created. If field does not
* exist or holds a string, the value is set to 0 before applying the
* operation. Since the value argument is signed you can use this command to
* perform both increments and decrements.
* precision floating point value. If key does not exist, a new key holding
* a hash is created. If field does not exist or holds a string, the value
* is set to 0 before applying the operation. Since the value argument is
* signed you can use this command to perform both increments and
* decrements.
* <p>
* The range of values supported by HINCRBYFLOAT is limited to
* double precision floating point values.
* The range of values supported by HINCRBYFLOAT is limited to double
* precision floating point values.
* <p>
* <b>Time complexity:</b> O(1)
*
*
* @param key
* @param field
* @param value
* @return Double precision floating point reply The new value at field after the increment
* operation.
* @return Double precision floating point reply The new value at field
* after the increment operation.
*/
public Double hincrByFloat(final String key, final String field, final double value) {
checkIsInMulti();
client.hincrByFloat(key, field, value);
final String dval = client.getBulkReply();
return (dval != null ? new Double(dval) : null);
public Double hincrByFloat(final String key, final String field,
final double value) {
checkIsInMulti();
client.hincrByFloat(key, field, value);
final String dval = client.getBulkReply();
return (dval != null ? new Double(dval) : null);
}
/**
@@ -1116,8 +1122,8 @@ public class Jedis extends BinaryJedis implements JedisCommands,
/**
* Atomically return and remove the first (LPOP) or last (RPOP) element of
* the list. For example if the list contains the elements "a","b","c" LPOP
* will return "a" and the list will become "b","c".
* the list. For example if the list contains the elements "a","b","c" RPOP
* will return "c" and the list will become "a","b".
* <p>
* If the key does not exist or the list is already empty the special value
* 'nil' is returned.
@@ -2609,6 +2615,35 @@ public class Jedis extends BinaryJedis implements JedisCommands,
client.zinterstore(dstkey, params, sets);
return client.getIntegerReply();
}
@Override
public Long zlexcount(final String key, final String min, final String max) {
checkIsInMulti();
client.zlexcount(key, min, max);
return client.getIntegerReply();
}
@Override
public Set<String> zrangeByLex(final String key, final String min, final String max) {
checkIsInMulti();
client.zrangeByLex(key, min, max);
return new LinkedHashSet<String>(client.getMultiBulkReply());
}
@Override
public Set<String> zrangeByLex(final String key, final String min, final String max,
final int offset, final int count) {
checkIsInMulti();
client.zrangeByLex(key, min, max, offset, count);
return new LinkedHashSet<String>(client.getMultiBulkReply());
}
@Override
public Long zremrangeByLex(final String key, final String min, final String max) {
checkIsInMulti();
client.zremrangeByLex(key, min, max);
return client.getIntegerReply();
}
public Long strlen(final String key) {
client.strlen(key);
@@ -2707,12 +2742,13 @@ public class Jedis extends BinaryJedis implements JedisCommands,
client.getrange(key, startOffset, endOffset);
return client.getBulkReply();
}
public Long bitpos(final String key, final boolean value) {
return bitpos(key, value, new BitPosParams());
}
public Long bitpos(final String key, final boolean value, final BitPosParams params) {
public Long bitpos(final String key, final boolean value,
final BitPosParams params) {
client.bitpos(key, value, params);
return client.getIntegerReply();
}
@@ -3117,11 +3153,6 @@ public class Jedis extends BinaryJedis implements JedisCommands,
return client.getStatusCodeReply();
}
@Deprecated
public Long pexpire(final String key, final int milliseconds) {
return pexpire(key, (long) milliseconds);
}
public Long pexpire(final String key, final long milliseconds) {
checkIsInMulti();
client.pexpire(key, milliseconds);
@@ -3140,7 +3171,6 @@ public class Jedis extends BinaryJedis implements JedisCommands,
return client.getIntegerReply();
}
public String psetex(final String key, final int milliseconds,
final String value) {
checkIsInMulti();
@@ -3180,131 +3210,6 @@ public class Jedis extends BinaryJedis implements JedisCommands,
return client.getStatusCodeReply();
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<String> scan(int cursor) {
return scan(cursor, new ScanParams());
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<String> scan(int cursor, final ScanParams params) {
checkIsInMulti();
client.scan(cursor, params);
List<Object> result = client.getObjectMultiBulkReply();
int newcursor = Integer.parseInt(new String((byte[]) result.get(0)));
List<String> results = new ArrayList<String>();
List<byte[]> rawResults = (List<byte[]>) result.get(1);
for (byte[] bs : rawResults) {
results.add(SafeEncoder.encode(bs));
}
return new ScanResult<String>(newcursor, results);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<Map.Entry<String, String>> hscan(final String key,
int cursor) {
return hscan(key, cursor, new ScanParams());
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<Map.Entry<String, String>> hscan(final String key,
int cursor, final ScanParams params) {
checkIsInMulti();
client.hscan(key, cursor, params);
List<Object> result = client.getObjectMultiBulkReply();
int newcursor = Integer.parseInt(new String((byte[]) result.get(0)));
List<Map.Entry<String, String>> results = new ArrayList<Map.Entry<String, String>>();
List<byte[]> rawResults = (List<byte[]>) result.get(1);
Iterator<byte[]> iterator = rawResults.iterator();
while (iterator.hasNext()) {
results.add(new AbstractMap.SimpleEntry<String, String>(SafeEncoder
.encode(iterator.next()), SafeEncoder.encode(iterator
.next())));
}
return new ScanResult<Map.Entry<String, String>>(newcursor, results);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<String> sscan(final String key, int cursor) {
return sscan(key, cursor, new ScanParams());
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<String> sscan(final String key, int cursor,
final ScanParams params) {
checkIsInMulti();
client.sscan(key, cursor, params);
List<Object> result = client.getObjectMultiBulkReply();
int newcursor = Integer.parseInt(new String((byte[]) result.get(0)));
List<String> results = new ArrayList<String>();
List<byte[]> rawResults = (List<byte[]>) result.get(1);
for (byte[] bs : rawResults) {
results.add(SafeEncoder.encode(bs));
}
return new ScanResult<String>(newcursor, results);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<Tuple> zscan(final String key, int cursor) {
return zscan(key, cursor, new ScanParams());
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<Tuple> zscan(final String key, int cursor,
final ScanParams params) {
checkIsInMulti();
client.zscan(key, cursor, params);
List<Object> result = client.getObjectMultiBulkReply();
int newcursor = Integer.parseInt(new String((byte[]) result.get(0)));
List<Tuple> results = new ArrayList<Tuple>();
List<byte[]> rawResults = (List<byte[]>) result.get(1);
Iterator<byte[]> iterator = rawResults.iterator();
while (iterator.hasNext()) {
results.add(new Tuple(SafeEncoder.encode(iterator.next()), Double
.valueOf(SafeEncoder.encode(iterator.next()))));
}
return new ScanResult<Tuple>(newcursor, results);
}
public ScanResult<String> scan(final String cursor) {
return scan(cursor, new ScanParams());
}
@@ -3435,60 +3340,67 @@ public class Jedis extends BinaryJedis implements JedisCommands,
client.clusterSetSlotImporting(slot, nodeId);
return client.getStatusCodeReply();
}
public String clusterSetSlotStable(final int slot) {
checkIsInMulti();
client.clusterSetSlotStable(slot);
return client.getStatusCodeReply();
}
public String clusterForget(final String nodeId) {
checkIsInMulti();
client.clusterForget(nodeId);
return client.getStatusCodeReply();
}
public String clusterFlushSlots() {
checkIsInMulti();
client.clusterFlushSlots();
return client.getStatusCodeReply();
}
public Long clusterKeySlot(final String key) {
checkIsInMulti();
client.clusterKeySlot(key);
return client.getIntegerReply();
}
public Long clusterCountKeysInSlot(final int slot) {
checkIsInMulti();
client.clusterCountKeysInSlot(slot);
return client.getIntegerReply();
}
public String clusterSaveConfig() {
checkIsInMulti();
client.clusterSaveConfig();
return client.getStatusCodeReply();
}
public String clusterReplicate(final String nodeId) {
checkIsInMulti();
client.clusterReplicate(nodeId);
return client.getStatusCodeReply();
}
public List<String> clusterSlaves(final String nodeId) {
checkIsInMulti();
client.clusterSlaves(nodeId);
return client.getMultiBulkReply();
}
public String clusterFailover() {
checkIsInMulti();
client.clusterFailover();
return client.getStatusCodeReply();
}
@Override
public List<Object> clusterSlots() {
checkIsInMulti();
client.clusterSlots();
return client.getObjectMultiBulkReply();
}
public String asking() {
checkIsInMulti();
@@ -3511,7 +3423,7 @@ public class Jedis extends BinaryJedis implements JedisCommands,
public Map<String, String> pubsubNumSub(String... channels) {
checkIsInMulti();
client.pubsubNumSub(channels);
return BuilderFactory.STRING_MAP
return BuilderFactory.PUBSUB_NUMSUB_MAP
.build(client.getBinaryMultiBulkReply());
}
@@ -3529,7 +3441,7 @@ public class Jedis extends BinaryJedis implements JedisCommands,
}
public void setDataSource(Pool<Jedis> jedisPool) {
this.dataSource = jedisPool;
this.dataSource = jedisPool;
}
public Long pfadd(final String key, final String... elements) {
@@ -3546,7 +3458,7 @@ public class Jedis extends BinaryJedis implements JedisCommands,
@Override
public long pfcount(String... keys) {
checkIsInMulti();
checkIsInMulti();
client.pfcount(keys);
return client.getIntegerReply();
}
@@ -3556,4 +3468,31 @@ public class Jedis extends BinaryJedis implements JedisCommands,
client.pfmerge(destkey, sourcekeys);
return client.getStatusCodeReply();
}
@Override
public List<String> blpop(int timeout, String key) {
checkIsInMulti();
List<String> args = new ArrayList<String>();
args.add(key);
args.add(String.valueOf(timeout));
client.blpop(args.toArray(new String[args.size()]));
client.setTimeoutInfinite();
final List<String> multiBulkReply = client.getMultiBulkReply();
client.rollbackTimeout();
return multiBulkReply;
}
@Override
public List<String> brpop(int timeout, String key) {
checkIsInMulti();
List<String> args = new ArrayList<String>();
args.add(key);
args.add(String.valueOf(timeout));
client.brpop(args.toArray(new String[args.size()]));
client.setTimeoutInfinite();
final List<String> multiBulkReply = client.getMultiBulkReply();
client.rollbackTimeout();
return multiBulkReply;
}
}

150
src/main/java/redis/clients/jedis/JedisCluster.java
View File

@@ -2,6 +2,7 @@ package redis.clients.jedis;
import redis.clients.jedis.BinaryClient.LIST_POSITION;
import java.io.Closeable;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -9,7 +10,7 @@ import java.util.Set;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
public class JedisCluster implements JedisCommands, BasicCommands {
public class JedisCluster implements JedisCommands, BasicCommands, Closeable {
public static final short HASHSLOTS = 16384;
private static final int DEFAULT_TIMEOUT = 1;
private static final int DEFAULT_MAX_REDIRECTIONS = 5;
@@ -50,6 +51,21 @@ public class JedisCluster implements JedisCommands, BasicCommands {
this.timeout = timeout;
this.maxRedirections = maxRedirections;
}
@Override
public void close() {
if (connectionHandler != null) {
for (JedisPool pool : connectionHandler.getNodes().values()) {
try {
if (pool != null) {
pool.destroy();
}
} catch (Exception e) {
// pass
}
}
}
}
@Override
public String set(final String key, final String value) {
@@ -412,7 +428,7 @@ public class JedisCluster implements JedisCommands, BasicCommands {
maxRedirections) {
@Override
public Long execute(Jedis connection) {
return connection.hdel(key);
return connection.hlen(key);
}
}.run(key);
}
@@ -641,6 +657,17 @@ public class JedisCluster implements JedisCommands, BasicCommands {
}.run(key);
}
@Override
public List<String> srandmember(final String key, final int count) {
return new JedisClusterCommand<List<String>>(connectionHandler, timeout,
maxRedirections) {
@Override
public List<String> execute(Jedis connection) {
return connection.srandmember(key, count);
}
}.run(key);
}
@Override
public Long strlen(final String key) {
return new JedisClusterCommand<Long>(connectionHandler, timeout,
@@ -1093,6 +1120,51 @@ public class JedisCluster implements JedisCommands, BasicCommands {
}
}.run(key);
}
@Override
public Long zlexcount(final String key, final String min, final String max) {
return new JedisClusterCommand<Long>(connectionHandler, timeout,
maxRedirections) {
@Override
public Long execute(Jedis connection) {
return connection.zlexcount(key, min, max);
}
}.run(key);
}
@Override
public Set<String> zrangeByLex(final String key, final String min, final String max) {
return new JedisClusterCommand<Set<String>>(connectionHandler, timeout,
maxRedirections) {
@Override
public Set<String> execute(Jedis connection) {
return connection.zrangeByLex(key, min, max);
}
}.run(key);
}
@Override
public Set<String> zrangeByLex(final String key, final String min, final String max,
final int offset, final int count) {
return new JedisClusterCommand<Set<String>>(connectionHandler, timeout,
maxRedirections) {
@Override
public Set<String> execute(Jedis connection) {
return connection.zrangeByLex(key, min, max, offset, count);
}
}.run(key);
}
@Override
public Long zremrangeByLex(final String key, final String min, final String max) {
return new JedisClusterCommand<Long>(connectionHandler, timeout,
maxRedirections) {
@Override
public Long execute(Jedis connection) {
return connection.zremrangeByLex(key, min, max);
}
}.run(key);
}
@Override
public Long linsert(final String key, final LIST_POSITION where,
@@ -1426,58 +1498,6 @@ public class JedisCluster implements JedisCommands, BasicCommands {
return null;
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
@Override
public ScanResult<Entry<String, String>> hscan(final String key,
final int cursor) {
return new JedisClusterCommand<ScanResult<Entry<String, String>>>(
connectionHandler, timeout, maxRedirections) {
@Override
public ScanResult<Entry<String, String>> execute(Jedis connection) {
return connection.hscan(key, cursor);
}
}.run(null);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
@Override
public ScanResult<String> sscan(final String key, final int cursor) {
return new JedisClusterCommand<ScanResult<String>>(connectionHandler,
timeout, maxRedirections) {
@Override
public ScanResult<String> execute(Jedis connection) {
return connection.sscan(key, cursor);
}
}.run(null);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
@Override
public ScanResult<Tuple> zscan(final String key, final int cursor) {
return new JedisClusterCommand<ScanResult<Tuple>>(connectionHandler,
timeout, maxRedirections) {
@Override
public ScanResult<Tuple> execute(Jedis connection) {
return connection.zscan(key, cursor);
}
}.run(null);
}
@Override
public ScanResult<Entry<String, String>> hscan(final String key,
final String cursor) {
@@ -1533,4 +1553,26 @@ public class JedisCluster implements JedisCommands, BasicCommands {
}
}.run(key);
}
@Override
public List<String> blpop(final int timeout, final String key) {
return new JedisClusterCommand<List<String>>(connectionHandler,
timeout, maxRedirections) {
@Override
public List<String> execute(Jedis connection) {
return connection.blpop(timeout,key);
}
}.run(null);
}
@Override
public List<String> brpop(final int timeout, final String key) {
return new JedisClusterCommand<List<String>>(connectionHandler,
timeout, maxRedirections) {
@Override
public List<String> execute(Jedis connection) {
return connection.brpop(timeout,key);
}
}.run(null);
}
}

20
src/main/java/redis/clients/jedis/JedisClusterCommand.java
View File

@@ -63,33 +63,35 @@ public abstract class JedisClusterCommand<T> {
// maybe all connection is down
throw jce;
}
releaseConnection(connection, true);
connection = null;
// retry with random connection
return runWithRetries(key, redirections--, true, asking);
} catch (JedisRedirectionException jre) {
if (jre instanceof JedisAskDataException) {
asking = true;
this.connectionHandler.assignSlotToNode(jre.getSlot(),
jre.getTargetNode());
} else if (jre instanceof JedisMovedDataException) {
// TODO : In antirez's redis-rb-cluster implementation,
// it rebuilds cluster's slot and node cache
// it rebuilds cluster's slot cache
// recommended by Redis cluster specification
this.connectionHandler.renewSlotCache();
} else {
throw new JedisClusterException(jre);
}
this.connectionHandler.assignSlotToNode(jre.getSlot(),
jre.getTargetNode());
releaseConnection(connection, false);
connection = null;
return runWithRetries(key, redirections - 1, false, asking);
} finally {
releaseConnection(connection, false);
}
}
private void releaseConnection(Jedis connection, boolean broken) {
if (connection != null) {
if (broken) {

115
src/main/java/redis/clients/jedis/JedisClusterConnectionHandler.java
View File

@@ -1,131 +1,86 @@
package redis.clients.jedis;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import java.util.*;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.ClusterNodeInformation;
import redis.clients.util.ClusterNodeInformationParser;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import static redis.clients.jedis.JedisClusterInfoCache.getNodeKey;
public abstract class JedisClusterConnectionHandler {
public static ClusterNodeInformationParser nodeInfoParser = new ClusterNodeInformationParser();
protected Map<String, JedisPool> nodes = new HashMap<String, JedisPool>();
protected Map<Integer, JedisPool> slots = new HashMap<Integer, JedisPool>();
final protected GenericObjectPoolConfig poolConfig;
protected final JedisClusterInfoCache cache;
abstract Jedis getConnection();
protected void returnConnection(Jedis connection) {
nodes.get(getNodeKey(connection.getClient()))
.returnResource(connection);
public void returnConnection(Jedis connection) {
cache.getNode(getNodeKey(connection.getClient())).returnResource(
connection);
}
public void returnBrokenConnection(Jedis connection) {
nodes.get(getNodeKey(connection.getClient())).returnBrokenResource(
cache.getNode(getNodeKey(connection.getClient())).returnBrokenResource(
connection);
}
abstract Jedis getConnectionFromSlot(int slot);
public JedisClusterConnectionHandler(Set<HostAndPort> nodes, final GenericObjectPoolConfig poolConfig) {
this.poolConfig = poolConfig;
initializeSlotsCache(nodes);
this.cache = new JedisClusterInfoCache(poolConfig);
initializeSlotsCache(nodes, poolConfig);
}
public Map<String, JedisPool> getNodes() {
return nodes;
return cache.getNodes();
}
private void initializeSlotsCache(Set<HostAndPort> startNodes) {
public void assignSlotToNode(int slot, HostAndPort targetNode) {
cache.assignSlotToNode(slot, targetNode);
}
private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig) {
for (HostAndPort hostAndPort : startNodes) {
JedisPool jp = new JedisPool(poolConfig, hostAndPort.getHost(),
hostAndPort.getPort());
this.nodes.clear();
this.slots.clear();
Jedis jedis = null;
try {
jedis = jp.getResource();
discoverClusterNodesAndSlots(jedis);
cache.discoverClusterNodesAndSlots(jedis);
break;
} catch (JedisConnectionException e) {
if (jedis != null) {
jp.returnBrokenResource(jedis);
jedis = null;
}
// try next nodes
} finally {
if (jedis != null) {
jp.returnResource(jedis);
jedis.close();
}
}
}
for (HostAndPort node : startNodes) {
setNodeIfNotExist(node);
cache.setNodeIfNotExist(node);
}
}
private void discoverClusterNodesAndSlots(Jedis jedis) {
String localNodes = jedis.clusterNodes();
for (String nodeInfo : localNodes.split("\n")) {
ClusterNodeInformation clusterNodeInfo = nodeInfoParser.parse(
nodeInfo, new HostAndPort(jedis.getClient().getHost(),
jedis.getClient().getPort()));
HostAndPort targetNode = clusterNodeInfo.getNode();
setNodeIfNotExist(targetNode);
assignSlotsToNode(clusterNodeInfo.getAvailableSlots(), targetNode);
}
}
public void assignSlotToNode(int slot, HostAndPort targetNode) {
JedisPool targetPool = nodes.get(getNodeKey(targetNode));
if (targetPool == null) {
setNodeIfNotExist(targetNode);
targetPool = nodes.get(getNodeKey(targetNode));
}
slots.put(slot, targetPool);
}
public void assignSlotsToNode(List<Integer> targetSlots,
HostAndPort targetNode) {
JedisPool targetPool = nodes.get(getNodeKey(targetNode));
if (targetPool == null) {
setNodeIfNotExist(targetNode);
targetPool = nodes.get(getNodeKey(targetNode));
}
for (Integer slot : targetSlots) {
slots.put(slot, targetPool);
public void renewSlotCache() {
for (JedisPool jp : cache.getNodes().values()) {
Jedis jedis = null;
try {
jedis = jp.getResource();
cache.discoverClusterSlots(jedis);
break;
} finally {
if (jedis != null) {
jedis.close();
}
}
}
}
protected JedisPool getRandomConnection() {
Object[] nodeArray = nodes.values().toArray();
Object[] nodeArray = cache.getNodes().values().toArray();
return (JedisPool) (nodeArray[new Random().nextInt(nodeArray.length)]);
}
protected String getNodeKey(HostAndPort hnp) {
return hnp.getHost() + ":" + hnp.getPort();
}
protected String getNodeKey(Client client) {
return client.getHost() + ":" + client.getPort();
}
private void setNodeIfNotExist(HostAndPort node) {
String nodeKey = getNodeKey(node);
if (nodes.containsKey(nodeKey))
return;
JedisPool nodePool = new JedisPool(poolConfig, node.getHost(), node.getPort());
nodes.put(nodeKey, nodePool);
}
}

188
src/main/java/redis/clients/jedis/JedisClusterInfoCache.java Normal file
View File

@@ -0,0 +1,188 @@
package redis.clients.jedis;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.util.ClusterNodeInformation;
import redis.clients.util.ClusterNodeInformationParser;
import redis.clients.util.SafeEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class JedisClusterInfoCache {
public static final ClusterNodeInformationParser nodeInfoParser = new ClusterNodeInformationParser();
private Map<String, JedisPool> nodes = new HashMap<String, JedisPool>();
private Map<Integer, JedisPool> slots = new HashMap<Integer, JedisPool>();
private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
private final Lock r = rwl.readLock();
private final Lock w = rwl.writeLock();
private final GenericObjectPoolConfig poolConfig;
public JedisClusterInfoCache(final GenericObjectPoolConfig poolConfig) {
this.poolConfig = poolConfig;
}
public void discoverClusterNodesAndSlots(Jedis jedis) {
w.lock();
try {
this.nodes.clear();
this.slots.clear();
String localNodes = jedis.clusterNodes();
for (String nodeInfo : localNodes.split("\n")) {
ClusterNodeInformation clusterNodeInfo = nodeInfoParser.parse(
nodeInfo, new HostAndPort(jedis.getClient().getHost(),
jedis.getClient().getPort()));
HostAndPort targetNode = clusterNodeInfo.getNode();
setNodeIfNotExist(targetNode);
assignSlotsToNode(clusterNodeInfo.getAvailableSlots(),
targetNode);
}
} finally {
w.unlock();
}
}
public void discoverClusterSlots(Jedis jedis) {
w.lock();
try {
this.slots.clear();
List<Object> slots = jedis.clusterSlots();
for (Object slotInfoObj : slots) {
List<Object> slotInfo = (List<Object>) slotInfoObj;
if (slotInfo.size() <= 2) {
continue;
}
List<Integer> slotNums = getAssignedSlotArray(slotInfo);
// hostInfos
List<Object> hostInfos = (List<Object>) slotInfo.get(2);
if (hostInfos.size() <= 0) {
continue;
}
// at this time, we just use master, discard slave information
HostAndPort targetNode = generateHostAndPort(hostInfos);
setNodeIfNotExist(targetNode);
assignSlotsToNode(slotNums, targetNode);
}
} finally {
w.unlock();
}
}
private HostAndPort generateHostAndPort(List<Object> hostInfos) {
return new HostAndPort(SafeEncoder.encode((byte[]) hostInfos.get(0)),
((Long) hostInfos.get(1)).intValue());
}
public void setNodeIfNotExist(HostAndPort node) {
w.lock();
try {
String nodeKey = getNodeKey(node);
if (nodes.containsKey(nodeKey))
return;
JedisPool nodePool = new JedisPool(poolConfig, node.getHost(), node.getPort());
nodes.put(nodeKey, nodePool);
} finally {
w.unlock();
}
}
public void assignSlotToNode(int slot, HostAndPort targetNode) {
w.lock();
try {
JedisPool targetPool = nodes.get(getNodeKey(targetNode));
if (targetPool == null) {
setNodeIfNotExist(targetNode);
targetPool = nodes.get(getNodeKey(targetNode));
}
slots.put(slot, targetPool);
} finally {
w.unlock();
}
}
public synchronized void assignSlotsToNode(List<Integer> targetSlots,
HostAndPort targetNode) {
w.lock();
try {
JedisPool targetPool = nodes.get(getNodeKey(targetNode));
if (targetPool == null) {
setNodeIfNotExist(targetNode);
targetPool = nodes.get(getNodeKey(targetNode));
}
for (Integer slot : targetSlots) {
slots.put(slot, targetPool);
}
} finally {
w.unlock();
}
}
public synchronized JedisPool getNode(String nodeKey) {
r.lock();
try {
return nodes.get(nodeKey);
} finally {
r.unlock();
}
}
public synchronized JedisPool getSlotPool(int slot) {
r.lock();
try {
return slots.get(slot);
} finally {
r.unlock();
}
}
public synchronized Map<String, JedisPool> getNodes() {
r.lock();
try {
return new HashMap<String, JedisPool>(nodes);
} finally {
r.unlock();
}
}
public static String getNodeKey(HostAndPort hnp) {
return hnp.getHost() + ":" + hnp.getPort();
}
public static String getNodeKey(Client client) {
return client.getHost() + ":" + client.getPort();
}
public static String getNodeKey(Jedis jedis) {
return getNodeKey(jedis.getClient());
}
private List<Integer> getAssignedSlotArray(List<Object> slotInfo) {
List<Integer> slotNums = new ArrayList<Integer>();
for (int slot = ((Long) slotInfo.get(0)).intValue(); slot <= ((Long) slotInfo
.get(1)).intValue(); slot++) {
slotNums.add(slot);
}
return slotNums;
}
}

39
src/main/java/redis/clients/jedis/JedisCommands.java
View File

@@ -113,6 +113,8 @@ public interface JedisCommands {
String srandmember(String key);
List<String> srandmember(String key, int count);
Long strlen(String key);
Long zadd(String key, double score, String member);
@@ -192,6 +194,15 @@ public interface JedisCommands {
Long zremrangeByScore(String key, double start, double end);
Long zremrangeByScore(String key, String start, String end);
Long zlexcount(final String key, final String min, final String max);
Set<String> zrangeByLex(final String key, final String min, final String max);
Set<String> zrangeByLex(final String key, final String min, final String max,
final int offset, final int count);
Long zremrangeByLex(final String key, final String min, final String max);
Long linsert(String key, Client.LIST_POSITION where, String pivot,
String value);
@@ -201,8 +212,12 @@ public interface JedisCommands {
Long rpushx(String key, String... string);
List<String> blpop(String arg);
List<String> blpop(int timeout, String key);
List<String> brpop(String arg);
List<String> brpop(int timeout, String key);
Long del(String key);
@@ -214,30 +229,6 @@ public interface JedisCommands {
Long bitcount(final String key, long start, long end);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
ScanResult<Map.Entry<String, String>> hscan(final String key, int cursor);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
ScanResult<String> sscan(final String key, int cursor);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
ScanResult<Tuple> zscan(final String key, int cursor);
ScanResult<Map.Entry<String, String>> hscan(final String key, final String cursor);
ScanResult<String> sscan(final String key, final String cursor);

23
src/main/java/redis/clients/jedis/JedisFactory.java
View File

@@ -1,5 +1,7 @@
package redis.clients.jedis;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
@@ -8,8 +10,7 @@ import org.apache.commons.pool2.impl.DefaultPooledObject;
* PoolableObjectFactory custom impl.
*/
class JedisFactory implements PooledObjectFactory<Jedis> {
private final String host;
private final int port;
private final AtomicReference<HostAndPort> hostAndPort = new AtomicReference<HostAndPort>();
private final int timeout;
private final String password;
private final int database;
@@ -23,14 +24,17 @@ class JedisFactory implements PooledObjectFactory<Jedis> {
public JedisFactory(final String host, final int port, final int timeout,
final String password, final int database, final String clientName) {
super();
this.host = host;
this.port = port;
this.hostAndPort.set(new HostAndPort(host, port));
this.timeout = timeout;
this.password = password;
this.database = database;
this.clientName = clientName;
}
public void setHostAndPort(final HostAndPort hostAndPort) {
this.hostAndPort.set(hostAndPort);
}
@Override
public void activateObject(PooledObject<Jedis> pooledJedis)
throws Exception {
@@ -60,7 +64,8 @@ class JedisFactory implements PooledObjectFactory<Jedis> {
@Override
public PooledObject<Jedis> makeObject() throws Exception {
final Jedis jedis = new Jedis(this.host, this.port, this.timeout);
final HostAndPort hostAndPort = this.hostAndPort.get();
final Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort(), this.timeout);
jedis.connect();
if (null != this.password) {
@@ -86,7 +91,13 @@ class JedisFactory implements PooledObjectFactory<Jedis> {
public boolean validateObject(PooledObject<Jedis> pooledJedis) {
final BinaryJedis jedis = pooledJedis.getObject();
try {
return jedis.isConnected() && jedis.ping().equals("PONG");
HostAndPort hostAndPort = this.hostAndPort.get();
String connectionHost = jedis.getClient().getHost();
int connectionPort = jedis.getClient().getPort();
return hostAndPort.getHost().equals(connectionHost) && hostAndPort.getPort() == connectionPort &&
jedis.isConnected() && jedis.ping().equals("PONG");
} catch (final Exception e) {
return false;
}

41
src/main/java/redis/clients/jedis/JedisPool.java
View File

@@ -5,6 +5,7 @@ import java.net.URI;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.util.JedisURIHelper;
import redis.clients.util.Pool;
public class JedisPool extends Pool<Jedis> {
@@ -24,8 +25,12 @@ public class JedisPool extends Pool<Jedis> {
if (uri.getScheme() != null && uri.getScheme().equals("redis")) {
String h = uri.getHost();
int port = uri.getPort();
String password = uri.getUserInfo().split(":", 2)[1];
int database = Integer.parseInt(uri.getPath().split("/", 2)[1]);
String password = JedisURIHelper.getPassword(uri);
int database = 0;
Integer dbIndex = JedisURIHelper.getDBIndex(uri);
if (dbIndex != null) {
database = dbIndex.intValue();
}
this.internalPool = new GenericObjectPool<Jedis>(
new JedisFactory(h, port, Protocol.DEFAULT_TIMEOUT,
password, database, null),
@@ -39,13 +44,11 @@ public class JedisPool extends Pool<Jedis> {
}
public JedisPool(final URI uri) {
String h = uri.getHost();
int port = uri.getPort();
String password = uri.getUserInfo().split(":", 2)[1];
int database = Integer.parseInt(uri.getPath().split("/", 2)[1]);
this.internalPool = new GenericObjectPool<Jedis>(new JedisFactory(h,
port, Protocol.DEFAULT_TIMEOUT, password, database, null),
new GenericObjectPoolConfig());
this(new GenericObjectPoolConfig(), uri, Protocol.DEFAULT_TIMEOUT);
}
public JedisPool(final URI uri, final int timeout) {
this(new GenericObjectPoolConfig(), uri, timeout);
}
public JedisPool(final GenericObjectPoolConfig poolConfig,
@@ -79,6 +82,18 @@ public class JedisPool extends Pool<Jedis> {
database, clientName));
}
public JedisPool(final GenericObjectPoolConfig poolConfig, final URI uri) {
this(poolConfig, uri, Protocol.DEFAULT_TIMEOUT);
}
public JedisPool(final GenericObjectPoolConfig poolConfig, final URI uri,
final int timeout) {
super(poolConfig, new JedisFactory(uri.getHost(), uri.getPort(),
timeout, JedisURIHelper.getPassword(uri),
JedisURIHelper.getDBIndex(uri) != null ? JedisURIHelper
.getDBIndex(uri) : 0, null));
}
@Override
public Jedis getResource() {
Jedis jedis = super.getResource();
@@ -98,4 +113,12 @@ public class JedisPool extends Pool<Jedis> {
returnResourceObject(resource);
}
}
public int getNumActive() {
if (this.internalPool == null || this.internalPool.isClosed()) {
return -1;
}
return this.internalPool.getNumActive();
}
}

6
src/main/java/redis/clients/jedis/JedisPubSub.java
View File

@@ -162,12 +162,6 @@ public abstract class JedisPubSub {
/* Invalidate instance since this thread is no longer listening */
this.client = null;
/*
* Reset pipeline count because subscribe() calls would have increased
* it but nothing decremented it.
*/
client.resetPipelinedCount();
}
public int getSubscribedChannels() {

54
src/main/java/redis/clients/jedis/JedisSentinelPool.java
View File

@@ -65,6 +65,7 @@ public class JedisSentinelPool extends Pool<Jedis> {
public JedisSentinelPool(String masterName, Set<String> sentinels,
final GenericObjectPoolConfig poolConfig, int timeout,
final String password, final int database) {
this.poolConfig = poolConfig;
this.timeout = timeout;
this.password = password;
@@ -74,6 +75,7 @@ public class JedisSentinelPool extends Pool<Jedis> {
initPool(master);
}
private volatile JedisFactory factory;
private volatile HostAndPort currentHostMaster;
public void destroy() {
@@ -91,10 +93,18 @@ public class JedisSentinelPool extends Pool<Jedis> {
private void initPool(HostAndPort master) {
if (!master.equals(currentHostMaster)) {
currentHostMaster = master;
if (factory == null) {
factory = new JedisFactory(master.getHost(), master.getPort(),
timeout, password, database);
initPool(poolConfig, factory);
} else {
factory.setHostAndPort(currentHostMaster);
// although we clear the pool, we still have to check the returned object
// in getResource, this call only clears idle instances, not borrowed instances
internalPool.clear();
}
log.info("Created JedisPool to master at " + master);
initPool(poolConfig,
new JedisFactory(master.getHost(), master.getPort(),
timeout, password, database));
}
}
@@ -115,19 +125,30 @@ public class JedisSentinelPool extends Pool<Jedis> {
log.fine("Connecting to Sentinel " + hap);
Jedis jedis = null;
try {
Jedis jedis = new Jedis(hap.getHost(), hap.getPort());
jedis = new Jedis(hap.getHost(), hap.getPort());
if (master == null) {
master = toHostAndPort(jedis
.sentinelGetMasterAddrByName(masterName));
List<String> masterAddr = jedis
.sentinelGetMasterAddrByName(masterName);
if (masterAddr == null || masterAddr.size() != 2) {
log.warning("Can not get master addr, master name: "
+ masterName + ". Sentinel: " + hap + ".");
continue;
}
master = toHostAndPort(masterAddr);
log.fine("Found Redis master at " + master);
jedis.disconnect();
break outer;
}
} catch (JedisConnectionException e) {
log.warning("Cannot connect to sentinel running @ " + hap
+ ". Trying next one.");
} finally {
if (jedis != null) {
jedis.close();
}
}
}
@@ -164,9 +185,22 @@ public class JedisSentinelPool extends Pool<Jedis> {
@Override
public Jedis getResource() {
Jedis jedis = super.getResource();
jedis.setDataSource(this);
return jedis;
while (true) {
Jedis jedis = super.getResource();
jedis.setDataSource(this);
// get a reference because it can change concurrently
final HostAndPort master = currentHostMaster;
final HostAndPort connection = new HostAndPort(jedis.getClient().getHost(),
jedis.getClient().getPort());
if (master.equals(connection)) {
// connected to the correct master
return jedis;
} else {
returnBrokenResource(jedis);
}
}
}
public void returnBrokenResource(final Jedis resource) {

9
src/main/java/redis/clients/jedis/JedisSlotBasedConnectionHandler.java
View File

@@ -48,14 +48,9 @@ public class JedisSlotBasedConnectionHandler extends
throw new JedisConnectionException("no reachable node in cluster");
}
@Override
public void assignSlotToNode(int slot, HostAndPort targetNode) {
super.assignSlotToNode(slot, targetNode);
}
@Override
public Jedis getConnectionFromSlot(int slot) {
JedisPool connectionPool = slots.get(slot);
JedisPool connectionPool = cache.getSlotPool(slot);
if (connectionPool != null) {
// It can't guaranteed to get valid connection because of node assignment
return connectionPool.getResource();
@@ -66,7 +61,7 @@ public class JedisSlotBasedConnectionHandler extends
private List<JedisPool> getShuffledNodesPool() {
List<JedisPool> pools = new ArrayList<JedisPool>();
pools.addAll(nodes.values());
pools.addAll(cache.getNodes().values());
Collections.shuffle(pools);
return pools;
}

8
src/main/java/redis/clients/jedis/MultiKeyCommands.java
View File

@@ -70,14 +70,6 @@ public interface MultiKeyCommands {
Long bitop(BitOP op, final String destKey, String... srcKeys);
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
ScanResult<String> scan(int cursor);
ScanResult<String> scan(final String cursor);
String pfmerge(final String destkey, final String... sourcekeys);

29
src/main/java/redis/clients/jedis/Pipeline.java
View File

@@ -1,10 +1,10 @@
package redis.clients.jedis;
import redis.clients.jedis.exceptions.JedisDataException;
import java.util.ArrayList;
import java.util.List;
import redis.clients.jedis.exceptions.JedisDataException;
public class Pipeline extends MultiKeyPipelineBase {
private MultiResponseBuilder currentMulti;
@@ -26,7 +26,13 @@ public class Pipeline extends MultiKeyPipelineBase {
for (int i = 0; i < list.size(); i++) {
Response<?> response = responses.get(i);
response.set(list.get(i));
values.add(response.get());
Object builtResponse;
try {
builtResponse = response.get();
} catch (JedisDataException e) {
builtResponse = e;
}
values.add(builtResponse);
}
return values;
}
@@ -69,13 +75,25 @@ public class Pipeline extends MultiKeyPipelineBase {
return client;
}
public void clear() {
if (isInMulti()) {
discard();
}
sync();
}
public boolean isInMulti() {
return currentMulti != null;
}
/**
* Syncronize pipeline by reading all responses. This operation close the
* pipeline. In order to get return values from pipelined commands, capture
* the different Response<?> of the commands you execute.
*/
public void sync() {
List<Object> unformatted = client.getAll();
List<Object> unformatted = client.getMany(getPipelinedResponseLength());
for (Object o : unformatted) {
generateResponse(o);
}
@@ -90,7 +108,7 @@ public class Pipeline extends MultiKeyPipelineBase {
* @return A list of all the responses in the order you executed them.
*/
public List<Object> syncAndReturnAll() {
List<Object> unformatted = client.getAll();
List<Object> unformatted = client.getMany(getPipelinedResponseLength());
List<Object> formatted = new ArrayList<Object>();
for (Object o : unformatted) {
@@ -106,7 +124,6 @@ public class Pipeline extends MultiKeyPipelineBase {
public Response<String> discard() {
if (currentMulti == null)
throw new JedisDataException("DISCARD without MULTI");
client.discard();
currentMulti = null;
return getResponse(BuilderFactory.STRING);

64
src/main/java/redis/clients/jedis/PipelineBase.java
View File

@@ -177,9 +177,9 @@ abstract class PipelineBase extends Queable implements BinaryRedisPipeline,
return getResponse(BuilderFactory.BYTE_ARRAY);
}
public Response<Long> getrange(byte[] key, long startOffset, long endOffset) {
public Response<byte[]> getrange(byte[] key, long startOffset, long endOffset) {
getClient(key).getrange(key, startOffset, endOffset);
return getResponse(BuilderFactory.LONG);
return getResponse(BuilderFactory.BYTE_ARRAY);
}
public Response<Long> hdel(String key, String... field) {
@@ -1026,6 +1026,56 @@ abstract class PipelineBase extends Queable implements BinaryRedisPipeline,
return getResponse(BuilderFactory.DOUBLE);
}
@Override
public Response<Long> zlexcount(final byte[] key, final byte[] min, final byte[] max) {
getClient(key).zlexcount(key, min, max);
return getResponse(BuilderFactory.LONG);
}
@Override
public Response<Long> zlexcount(final String key, final String min, final String max) {
getClient(key).zlexcount(key, min, max);
return getResponse(BuilderFactory.LONG);
}
@Override
public Response<Set<byte[]>> zrangeByLex(final byte[] key, final byte[] max, final byte[] min) {
getClient(key).zrangeByLex(key, min, max);
return getResponse(BuilderFactory.BYTE_ARRAY_ZSET);
}
@Override
public Response<Set<String>> zrangeByLex(final String key, final String max, final String min) {
getClient(key).zrangeByLex(key, min, max);
return getResponse(BuilderFactory.STRING_ZSET);
}
@Override
public Response<Set<byte[]>> zrangeByLex(final byte[] key, final byte[] max,
final byte[] min, final int offset, final int count) {
getClient(key).zrangeByLex(key, min, max, offset, count);
return getResponse(BuilderFactory.BYTE_ARRAY_ZSET);
}
@Override
public Response<Set<String>> zrangeByLex(final String key, final String max,
final String min, final int offset, final int count) {
getClient(key).zrangeByLex(key, min, max, offset, count);
return getResponse(BuilderFactory.STRING_ZSET);
}
@Override
public Response<Long> zremrangeByLex(final byte[] key, final byte[] min, final byte[] max) {
getClient(key).zremrangeByLex(key, min, max);
return getResponse(BuilderFactory.LONG);
}
@Override
public Response<Long> zremrangeByLex(final String key, final String min, final String max) {
getClient(key).zremrangeByLex(key, min, max);
return getResponse(BuilderFactory.LONG);
}
public Response<Long> bitcount(String key) {
getClient(key).bitcount(key);
return getResponse(BuilderFactory.LONG);
@@ -1098,16 +1148,6 @@ abstract class PipelineBase extends Queable implements BinaryRedisPipeline,
return getResponse(BuilderFactory.LONG);
}
@Deprecated
public Response<Long> pexpire(String key, int milliseconds) {
return pexpire(key, (long) milliseconds);
}
@Deprecated
public Response<Long> pexpire(byte[] key, int milliseconds) {
return pexpire(key, (long) milliseconds);
}
public Response<Long> pexpire(String key, long milliseconds) {
getClient(key).pexpire(key, milliseconds);
return getResponse(BuilderFactory.LONG);

4
src/main/java/redis/clients/jedis/PipelineBlock.java
View File

@@ -7,5 +7,9 @@ package redis.clients.jedis;
* @see https://github.com/xetorthio/jedis/pull/498
*/
public abstract class PipelineBlock extends Pipeline {
// For shadowing
@SuppressWarnings("unused")
private Client client;
public abstract void execute();
}

3
src/main/java/redis/clients/jedis/Protocol.java
View File

@@ -59,6 +59,7 @@ public final class Protocol {
public static final String CLUSTER_REPLICATE = "replicate";
public static final String CLUSTER_SLAVES = "slaves";
public static final String CLUSTER_FAILOVER = "failover";
public static final String CLUSTER_SLOTS = "slots";
public static final String PUBSUB_CHANNELS= "channels";
public static final String PUBSUB_NUMSUB = "numsub";
public static final String PUBSUB_NUM_PAT = "numpat";
@@ -217,7 +218,7 @@ public final class Protocol {
}
public static enum Command {
PING, SET, GET, QUIT, EXISTS, DEL, TYPE, FLUSHDB, KEYS, RANDOMKEY, RENAME, RENAMENX, RENAMEX, DBSIZE, EXPIRE, EXPIREAT, TTL, SELECT, MOVE, FLUSHALL, GETSET, MGET, SETNX, SETEX, MSET, MSETNX, DECRBY, DECR, INCRBY, INCR, APPEND, SUBSTR, HSET, HGET, HSETNX, HMSET, HMGET, HINCRBY, HEXISTS, HDEL, HLEN, HKEYS, HVALS, HGETALL, RPUSH, LPUSH, LLEN, LRANGE, LTRIM, LINDEX, LSET, LREM, LPOP, RPOP, RPOPLPUSH, SADD, SMEMBERS, SREM, SPOP, SMOVE, SCARD, SISMEMBER, SINTER, SINTERSTORE, SUNION, SUNIONSTORE, SDIFF, SDIFFSTORE, SRANDMEMBER, ZADD, ZRANGE, ZREM, ZINCRBY, ZRANK, ZREVRANK, ZREVRANGE, ZCARD, ZSCORE, MULTI, DISCARD, EXEC, WATCH, UNWATCH, SORT, BLPOP, BRPOP, AUTH, SUBSCRIBE, PUBLISH, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PUBSUB, ZCOUNT, ZRANGEBYSCORE, ZREVRANGEBYSCORE, ZREMRANGEBYRANK, ZREMRANGEBYSCORE, ZUNIONSTORE, ZINTERSTORE, SAVE, BGSAVE, BGREWRITEAOF, LASTSAVE, SHUTDOWN, INFO, MONITOR, SLAVEOF, CONFIG, STRLEN, SYNC, LPUSHX, PERSIST, RPUSHX, ECHO, LINSERT, DEBUG, BRPOPLPUSH, SETBIT, GETBIT, BITPOS, SETRANGE, GETRANGE, EVAL, EVALSHA, SCRIPT, SLOWLOG, OBJECT, BITCOUNT, BITOP, SENTINEL, DUMP, RESTORE, PEXPIRE, PEXPIREAT, PTTL, INCRBYFLOAT, PSETEX, CLIENT, TIME, MIGRATE, HINCRBYFLOAT, SCAN, HSCAN, SSCAN, ZSCAN, WAIT, CLUSTER, ASKING, PFADD, PFCOUNT, PFMERGE;
PING, SET, GET, QUIT, EXISTS, DEL, TYPE, FLUSHDB, KEYS, RANDOMKEY, RENAME, RENAMENX, RENAMEX, DBSIZE, EXPIRE, EXPIREAT, TTL, SELECT, MOVE, FLUSHALL, GETSET, MGET, SETNX, SETEX, MSET, MSETNX, DECRBY, DECR, INCRBY, INCR, APPEND, SUBSTR, HSET, HGET, HSETNX, HMSET, HMGET, HINCRBY, HEXISTS, HDEL, HLEN, HKEYS, HVALS, HGETALL, RPUSH, LPUSH, LLEN, LRANGE, LTRIM, LINDEX, LSET, LREM, LPOP, RPOP, RPOPLPUSH, SADD, SMEMBERS, SREM, SPOP, SMOVE, SCARD, SISMEMBER, SINTER, SINTERSTORE, SUNION, SUNIONSTORE, SDIFF, SDIFFSTORE, SRANDMEMBER, ZADD, ZRANGE, ZREM, ZINCRBY, ZRANK, ZREVRANK, ZREVRANGE, ZCARD, ZSCORE, MULTI, DISCARD, EXEC, WATCH, UNWATCH, SORT, BLPOP, BRPOP, AUTH, SUBSCRIBE, PUBLISH, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PUBSUB, ZCOUNT, ZRANGEBYSCORE, ZREVRANGEBYSCORE, ZREMRANGEBYRANK, ZREMRANGEBYSCORE, ZUNIONSTORE, ZINTERSTORE, ZLEXCOUNT, ZRANGEBYLEX, ZREMRANGEBYLEX, SAVE, BGSAVE, BGREWRITEAOF, LASTSAVE, SHUTDOWN, INFO, MONITOR, SLAVEOF, CONFIG, STRLEN, SYNC, LPUSHX, PERSIST, RPUSHX, ECHO, LINSERT, DEBUG, BRPOPLPUSH, SETBIT, GETBIT, BITPOS, SETRANGE, GETRANGE, EVAL, EVALSHA, SCRIPT, SLOWLOG, OBJECT, BITCOUNT, BITOP, SENTINEL, DUMP, RESTORE, PEXPIRE, PEXPIREAT, PTTL, INCRBYFLOAT, PSETEX, CLIENT, TIME, MIGRATE, HINCRBYFLOAT, SCAN, HSCAN, SSCAN, ZSCAN, WAIT, CLUSTER, ASKING, PFADD, PFCOUNT, PFMERGE;
public final byte[] raw;

7
src/main/java/redis/clients/jedis/Queable.java
View File

@@ -24,4 +24,11 @@ public class Queable {
return lr;
}
protected boolean hasPipelinedResponse() {
return pipelinedResponses.size() > 0;
}
protected int getPipelinedResponseLength() {
return pipelinedResponses.size();
}
}

9
src/main/java/redis/clients/jedis/RedisPipeline.java
View File

@@ -184,6 +184,15 @@ public interface RedisPipeline {
Response<Long> zrevrank(String key, String member);
Response<Double> zscore(String key, String member);
Response<Long> zlexcount(final String key, final String min, final String max);
Response<Set<String>> zrangeByLex(final String key, final String max, final String min);
Response<Set<String>> zrangeByLex(final String key, final String max, final String min,
final int offset, final int count);
Response<Long> zremrangeByLex(final String key, final String start, final String end);
Response<Long> bitcount(String key);

26
src/main/java/redis/clients/jedis/ScanResult.java
View File

@@ -8,16 +8,6 @@ public class ScanResult<T> {
private byte[] cursor;
private List<T> results;
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult(int cursor, List<T> results) {
this(Protocol.toByteArray(cursor), results);
}
public ScanResult(String cursor, List<T> results) {
this(SafeEncoder.encode(cursor), results);
}
@@ -27,21 +17,7 @@ public class ScanResult<T> {
this.results = results;
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
* @return int(currently), but will be changed to String, so be careful to prepare!
*/
public int getCursor() {
return Integer.parseInt(getStringCursor());
}
/**
* FIXME: This method should be changed to getCursor() on next major release
*/
public String getStringCursor() {
public String getCursor() {
return SafeEncoder.encode(cursor);
}

74
src/main/java/redis/clients/jedis/ShardedJedis.java
View File

@@ -125,12 +125,18 @@ public class ShardedJedis extends BinaryShardedJedis implements JedisCommands,
Jedis j = getShard(arg);
return j.blpop(arg);
}
public List<String> blpop(int timeout,String key){
Jedis j = getShard(key);
return j.blpop(timeout,key);
}
public List<String> brpop(String arg) {
Jedis j = getShard(arg);
return j.brpop(arg);
}
public List<String> brpop(int timeout,String key) {
Jedis j = getShard(key);
return j.brpop(timeout,key);
}
public Long decrBy(String key, long integer) {
Jedis j = getShard(key);
return j.decrBy(key, integer);
@@ -346,6 +352,12 @@ public class ShardedJedis extends BinaryShardedJedis implements JedisCommands,
return j.srandmember(key);
}
@Override
public List<String> srandmember(String key, int count) {
Jedis j = getShard(key);
return j.srandmember(key, count);
}
public Long zadd(String key, double score, String member) {
Jedis j = getShard(key);
return j.zadd(key, score, member);
@@ -530,6 +542,27 @@ public class ShardedJedis extends BinaryShardedJedis implements JedisCommands,
Jedis j = getShard(key);
return j.zremrangeByScore(key, start, end);
}
@Override
public Long zlexcount(final String key, final String min, final String max) {
return getShard(key).zlexcount(key, min, max);
}
@Override
public Set<String> zrangeByLex(final String key, final String min, final String max) {
return getShard(key).zrangeByLex(key, min, max);
}
@Override
public Set<String> zrangeByLex(final String key, final String min, final String max,
final int offset, final int count) {
return getShard(key).zrangeByLex(key, min, max, offset, count);
}
@Override
public Long zremrangeByLex(final String key, final String min, final String max) {
return getShard(key).zremrangeByLex(key, min, max);
}
public Long linsert(String key, LIST_POSITION where, String pivot,
String value) {
@@ -547,41 +580,7 @@ public class ShardedJedis extends BinaryShardedJedis implements JedisCommands,
return j.bitcount(key, start, end);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<Entry<String, String>> hscan(String key, int cursor) {
Jedis j = getShard(key);
return j.hscan(key, cursor);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<String> sscan(String key, int cursor) {
Jedis j = getShard(key);
return j.sscan(key, cursor);
}
@Deprecated
/**
* This method is deprecated due to bug (scan cursor should be unsigned long)
* And will be removed on next major release
* @see https://github.com/xetorthio/jedis/issues/531
*/
public ScanResult<Tuple> zscan(String key, int cursor) {
Jedis j = getShard(key);
return j.zscan(key, cursor);
}
public ScanResult<Entry<String, String>> hscan(String key,
final String cursor) {
public ScanResult<Entry<String, String>> hscan(String key, final String cursor) {
Jedis j = getShard(key);
return j.hscan(key, cursor);
}
@@ -639,4 +638,5 @@ public class ShardedJedis extends BinaryShardedJedis implements JedisCommands,
Jedis j = getShard(key);
return j.pfcount(key);
}
}

18
src/main/java/redis/clients/jedis/Transaction.java
View File

@@ -31,9 +31,16 @@ public class Transaction extends MultiKeyPipelineBase {
return client;
}
public void clear() {
if (inTransaction) {
discard();
}
}
public List<Object> exec() {
// Discard QUEUED or ERROR
client.getMany(getPipelinedResponseLength());
client.exec();
client.getAll(1); // Discard all but the last reply
List<Object> unformatted = client.getObjectMultiBulkReply();
if (unformatted == null) {
@@ -51,8 +58,9 @@ public class Transaction extends MultiKeyPipelineBase {
}
public List<Response<?>> execGetResponse() {
// Discard QUEUED or ERROR
client.getMany(getPipelinedResponseLength());
client.exec();
client.getAll(1); // Discard all but the last reply
List<Object> unformatted = client.getObjectMultiBulkReply();
if (unformatted == null) {
@@ -66,11 +74,15 @@ public class Transaction extends MultiKeyPipelineBase {
}
public String discard() {
client.getMany(getPipelinedResponseLength());
client.discard();
client.getAll(1); // Discard all but the last reply
inTransaction = false;
clean();
return client.getStatusCodeReply();
}
public void setClient(Client client) {
this.client = client;
}
}

6
src/main/java/redis/clients/jedis/TransactionBlock.java
View File

@@ -9,6 +9,10 @@ import redis.clients.jedis.exceptions.JedisException;
* @see https://github.com/xetorthio/jedis/pull/498
*/
public abstract class TransactionBlock extends Transaction {
// For shadowing
@SuppressWarnings("unused")
private Client client;
public TransactionBlock(Client client) {
super(client);
}
@@ -19,6 +23,6 @@ public abstract class TransactionBlock extends Transaction {
public abstract void execute() throws JedisException;
public void setClient(Client client) {
this.client = client;
super.setClient(client);
}
}

10
src/main/java/redis/clients/jedis/ZParams.java
View File

@@ -23,9 +23,15 @@ public class ZParams {
private List<byte[]> params = new ArrayList<byte[]>();
public ZParams weights(final int... weights) {
/**
* Set weights.
*
* @param weights
* weights.
*/
public ZParams weights(final double... weights) {
params.add(WEIGHTS.raw);
for (final int weight : weights) {
for (final double weight : weights) {
params.add(Protocol.toByteArray(weight));
}

11
src/main/java/redis/clients/util/ClusterNodeInformationParser.java
View File

@@ -3,7 +3,6 @@ package redis.clients.util;
import redis.clients.jedis.HostAndPort;
public class ClusterNodeInformationParser {
private static final String HOST_MYSELF_IDENTIFIER = ":0";
private static final String SLOT_IMPORT_IDENTIFIER = "-<-";
private static final String SLOT_IN_TRANSITION_IDENTIFIER = "[";
public static final int SLOT_INFORMATIONS_START_INDEX = 8;
@@ -36,13 +35,13 @@ public class ClusterNodeInformationParser {
public HostAndPort getHostAndPortFromNodeLine(String[] nodeInfoPartArray,
HostAndPort current) {
String stringHostAndPort = nodeInfoPartArray[HOST_AND_PORT_INDEX];
if (HOST_MYSELF_IDENTIFIER.equals(stringHostAndPort)) {
return current;
}
String[] arrayHostAndPort = stringHostAndPort.split(":");
return new HostAndPort(arrayHostAndPort[0],
Integer.valueOf(arrayHostAndPort[1]));
return new HostAndPort(
arrayHostAndPort[0].isEmpty() ? current.getHost()
: arrayHostAndPort[0],
arrayHostAndPort[1].isEmpty() ? current.getPort() : Integer
.valueOf(arrayHostAndPort[1]));
}
private void fillSlotInformation(String[] slotInfoPartArray,

26
src/main/java/redis/clients/util/JedisURIHelper.java Normal file
View File

@@ -0,0 +1,26 @@
package redis.clients.util;
import java.net.URI;
public class JedisURIHelper {
public static String getPassword(URI uri) {
String userInfo = uri.getUserInfo();
if (userInfo != null) {
return userInfo.split(":", 2)[1];
}
return null;
}
public static Integer getDBIndex(URI uri) {
String[] pathSplit = uri.getPath().split("/", 2);
if (pathSplit.length > 1) {
String dbIndexStr = pathSplit[1];
if (dbIndexStr.isEmpty()) {
return 0;
}
return Integer.parseInt(dbIndexStr);
} else {
return 0;
}
}
}

13
src/main/java/redis/clients/util/Pool.java
View File

@@ -1,5 +1,7 @@
package redis.clients.util;
import java.io.Closeable;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
@@ -7,7 +9,7 @@ import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;
public abstract class Pool<T> {
public abstract class Pool<T> implements Closeable {
protected GenericObjectPool<T> internalPool;
/**
@@ -17,6 +19,15 @@ public abstract class Pool<T> {
public Pool() {
}
@Override
public void close() {
closeInternalPool();
}
public boolean isClosed() {
return this.internalPool.isClosed();
}
public Pool(final GenericObjectPoolConfig poolConfig,
PooledObjectFactory<T> factory) {
initPool(poolConfig, factory);