/*
* Copyright 2009-2010 Joubin Houshyar
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.jredis.ri.alphazero;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.StringTokenizer;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.jredis.ClientRuntimeException;
import org.jredis.JRedisFuture;
import org.jredis.KeyValueSet;
import org.jredis.ObjectInfo;
import org.jredis.ProviderException;
import org.jredis.RedisType;
import org.jredis.Sort;
import org.jredis.ZSetEntry;
import org.jredis.connector.Connection;
import org.jredis.protocol.BulkResponse;
import org.jredis.protocol.Command;
import org.jredis.protocol.MultiBulkResponse;
import org.jredis.protocol.Response;
import org.jredis.protocol.ResponseStatus;
import org.jredis.protocol.ValueResponse;
import org.jredis.ri.alphazero.support.Convert;
import org.jredis.ri.alphazero.support.DefaultCodec;
import org.jredis.ri.alphazero.support.SortSupport;
/**
*
* [TODO: document me!]
*
* @author Joubin Houshyar (alphazero@sensesay.net)
* @version alpha.0, Apr 10, 2009
* @since alpha.0
*
*/
public abstract class JRedisFutureSupport implements JRedisFuture {
// ------------------------------------------------------------------------
// Properties
// ------------------------------------------------------------------------
// ------------------------------------------------------------------------
// Constructors
// ------------------------------------------------------------------------
// ------------------------------------------------------------------------
// Extension point(s)
/*
* This class provides the convenience of a uniform implementation wide mapping
* of JRedis api semantics to the native protocol level semantics of byte[]s.
*
* Extensions can use the provided extension points to provide or delegate the
* servicing of request calls.
*/
// ------------------------------------------------------------------------
/**
* This method mimics the eponymous {@link Connection#queueRequest(Command, byte[]...)}
* which defines the blocking api semantics of Synchronous connections. The extending class
* can either directly (a) implement the protocol requirements, or, (b) delegate to a
* {@link Connection} instance, or, (c) utilize a pool of {@link Connection}s.
*
* @param cmd
* @param args
* @return
* @throws ClientRuntimeException
* @throws ProviderException
*/
abstract protected Future<Response> queueRequest (Command cmd, byte[]...args) throws ClientRuntimeException, ProviderException;
// ------------------------------------------------------------------------
// INTERFACE
// ================================================================ Redis
/*
* Support of all the JRedisFuture interface methods.
*
* This class uses the UTF-8 character set for all conversions due to its
* use of the Convert and Codec support classes.
*
* All calls are forwarded to an abstract queueRequest method that the
* extending classes are expected to implement.
*/
// ------------------------------------------------------------------------
@Override
public <K extends Object> FutureStatus bgsave() {
return new FutureStatus(this.queueRequest(Command.BGSAVE));
}
@Override
public <K extends Object> FutureString bgrewriteaof() {
Future<Response> futureResponse = this.queueRequest(Command.BGREWRITEAOF);
return new FutureString(futureResponse);
}
@Override
public <K extends Object> FutureStatus ping() {
return new FutureStatus(this.queueRequest(Command.PING));
}
@Override
public <K extends Object> FutureStatus flushall() {
return new FutureStatus(this.queueRequest(Command.FLUSHALL));
}
@Override
public <K extends Object> FutureStatus flushdb() {
return new FutureStatus(this.queueRequest(Command.FLUSHDB));
}
// public <K extends Object> FutureStatus select(int index) {
// this.queueRequest(Command.SELECT, Convert.toBytes(index));
// return this;
// }
@Override
public <K extends Object> Future<ResponseStatus> slaveof(String host, int port) {
byte[] hostbytes = null;
if((hostbytes = JRedisSupport.getKeyBytes(host)) == null)
throw new IllegalArgumentException ("invalid host => ["+host+"]");
byte[] portbytes = null;
if((portbytes = Convert.toBytes(port)) == null)
throw new IllegalArgumentException ("invalid port => ["+port+"]");
return new FutureStatus(this.queueRequest(Command.SLAVEOF, hostbytes, portbytes));
}
public <K extends Object> Future<ResponseStatus> slaveofnone() {
return new FutureStatus(this.queueRequest(Command.SLAVEOF, "no".getBytes(), "one".getBytes()));
}
@Override
public <K extends Object> FutureStatus rename(K oldkey, K newkey) {
byte[] oldkeydata = null;
if((oldkeydata = JRedisSupport.getKeyBytes(oldkey)) == null)
throw new IllegalArgumentException ("invalid key => ["+oldkey+"]");
byte[] newkeydata = null;
if((newkeydata = JRedisSupport.getKeyBytes(newkey)) == null)
throw new IllegalArgumentException ("invalid key => ["+newkey+"]");
return new FutureStatus(this.queueRequest(Command.RENAME, oldkeydata, newkeydata));
}
@Override
public <K extends Object> Future<Boolean> renamenx(K oldkey, K newkey){
byte[] oldkeydata = null;
if((oldkeydata = JRedisSupport.getKeyBytes(oldkey)) == null)
throw new IllegalArgumentException ("invalid key => ["+oldkey+"]");
byte[] newkeydata = null;
if((newkeydata = JRedisSupport.getKeyBytes(newkey)) == null)
throw new IllegalArgumentException ("invalid key => ["+newkey+"]");
Future<Response> futureResponse = this.queueRequest(Command.RENAMENX, oldkeydata, newkeydata);
return new FutureBoolean(futureResponse);
}
public <K extends Object> FutureLong rpush(K key, byte[] value) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
if(value == null)
throw new IllegalArgumentException ("null value");
return new FutureLong(this.queueRequest(Command.RPUSH, keybytes, value));
}
public <K extends Object> FutureLong rpushx(K key, byte[] value) {
byte[] keybytes = null;
if ((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
if (value == null)
throw new IllegalArgumentException ("null value");
return new FutureLong(this.queueRequest(Command.RPUSHX, keybytes, value));
}
public <K extends Object> FutureLong lpushx(K key, byte[] value) {
byte[] keybytes = null;
if ((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
if (value == null)
throw new IllegalArgumentException ("null value");
return new FutureLong(this.queueRequest(Command.LPUSHX, keybytes, value));
}
public <K extends Object> FutureLong linsert(K key, boolean after, byte[] oldvalue, byte[] newvalue) {
byte[] keybytes = null;
if ((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[][] bulk = new byte[4][];
bulk[0] = keybytes;
bulk[1] = (after ? "AFTER" : "BEFORE").getBytes();
bulk[2] = oldvalue;
bulk[3] = newvalue;
return new FutureLong(this.queueRequest(Command.LINSERT, bulk));
}
public <K extends Object> FutureLong linsertAfter(K key, byte[] oldvalue, byte[] newvalue) {
return linsert(key, true, oldvalue, newvalue);
}
public <K extends Object> FutureLong linsertBefore(K key, byte[] oldvalue, byte[] newvalue) {
return linsert(key, false, oldvalue, newvalue);
}
@Override
public <K extends Object> FutureByteArray rpoplpush (String srcList, String destList) {
byte[] srckeybytes = null;
if((srckeybytes = JRedisSupport.getKeyBytes(srcList)) == null)
throw new IllegalArgumentException ("invalid src key => ["+srcList+"]");
byte[] destkeybytes = null;
if((destkeybytes = JRedisSupport.getKeyBytes(destList)) == null)
throw new IllegalArgumentException ("invalid dest key => ["+destList+"]");
Future<Response> futureResponse = this.queueRequest(Command.RPOPLPUSH, srckeybytes, destkeybytes);
return new FutureByteArray(futureResponse);
}
@Override
public <K extends Object> FutureLong rpush(K key, String value) {
// rpush(key, DefaultCodec.encode(value));
return rpush(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> FutureLong rpush(K key, Number value) {
return rpush(key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> FutureLong rpush (K key, T value)
{
return rpush(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> sadd(K key, byte[] member)
{
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SADD, keybytes, member);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> sadd (K key, String value) {
return sadd (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> sadd (K key, Number value) {
return sadd (key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Boolean> sadd (K key, T value)
{
return sadd (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> zadd(K key, double score, byte[] member)
{
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZADD, keybytes, Convert.toBytes(score), member);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> zadd (K key, double score, String value) {
return zadd (key, score, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> zadd (K key, double score, Number value) {
return zadd (key, score, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Boolean> zadd (K key, double score, T value)
{
return zadd (key, score, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Double> zincrby(K key, double score, byte[] member)
{
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZINCRBY, keybytes, Convert.toBytes(score), member);
return new FutureDouble(futureResponse);
}
@Override
public <K extends Object> Future<Double> zincrby (K key, double score, String value) {
return zincrby (key, score, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Double> zincrby (K key, double score, Number value) {
return zincrby (key, score, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Double> zincrby (K key, double score, T value)
{
return zincrby (key, score, DefaultCodec.encode(value));
}
@Override
public <K extends Object> FutureStatus save()
{
return new FutureStatus(this.queueRequest(Command.SAVE));
}
// -------- set
@Override
public <K> Future<Boolean> setbit(K key, int offset, boolean value) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SETBIT, keybytes,
Convert.toBytes(offset), Convert.toBytes(value ? 1 : 0));
return new FutureBit(futureResponse);
}
@Override
public <K> Future<Boolean> getbit(K key, int offset) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.GETBIT, keybytes,
Convert.toBytes(offset));
return new FutureBit(futureResponse);
}
@Override
public <K extends Object> FutureStatus set(K key, byte[] value) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
return new FutureStatus(this.queueRequest(Command.SET, keybytes, value));
}
@Override
public <K extends Object> FutureStatus set(K key, String value) {
return set(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> FutureStatus set(K key, Number value) {
return set(key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> FutureStatus set (K key, T value)
{
return set(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<byte[]> getset(K key, byte[] value) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.GETSET, keybytes, value);
return new FutureByteArray(futureResponse);
}
@Override
public <K extends Object> Future<byte[]> getset(K key, String value) {
return getset(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<byte[]> getset(K key, Number value) {
return getset(key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable>
Future<byte[]> getset (K key, T value)
{
return getset(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> setnx(K key, byte[] value){
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SETNX, keybytes, value);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> setnx(K key, String value) {
return setnx(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> setnx(K key, Number value) {
return setnx(key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Boolean> setnx (K key, T value) {
return setnx(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> append (K key, byte[] value){
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.APPEND, keybytes, value);
return new FutureLong(futureResponse);
}
@Override
public <K extends Object> Future<Long> append(K key, String value) {
return append(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> append(K key, Number value) {
return append(key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Long> append (K key, T value) {
return append(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> sismember(K key, byte[] member) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SISMEMBER, keybytes, member);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> sismember(K key, String value) {
return sismember(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> sismember(K key, Number numberValue) {
return sismember (key, String.valueOf(numberValue).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Boolean> sismember(K key, T object) {
return sismember(key, DefaultCodec.encode(object));
}
public <K extends Object> Future<Boolean> smove (K srcKey, K destKey, byte[] member) {
byte[] srcKeyBytes = null;
if((srcKeyBytes = JRedisSupport.getKeyBytes(srcKey)) == null)
throw new IllegalArgumentException ("invalid key => ["+srcKey+"]");
byte[] destKeyBytes = null;
if((destKeyBytes = JRedisSupport.getKeyBytes(destKey)) == null)
throw new IllegalArgumentException ("invalid key => ["+destKey+"]");
Future<Response> futureResponse = this.queueRequest(Command.SMOVE, srcKeyBytes, destKeyBytes, member);
return new FutureBoolean(futureResponse);
}
public <K extends Object> Future<Boolean> smove (K srcKey, K destKey, String stringValue) {
return smove (srcKey, destKey, DefaultCodec.encode(stringValue));
}
public <K extends Object> Future<Boolean> smove (K srcKey, K destKey, Number numberValue) {
return smove (srcKey, destKey, String.valueOf(numberValue).getBytes());
}
public <K extends Object, T extends Serializable>
Future<Boolean> smove (K srcKey, K destKey, T object) {
return smove (srcKey, destKey, DefaultCodec.encode(object));
}
// ------------------------------------------------------------------------
// Commands operating on hashes
// ------------------------------------------------------------------------
public <K extends Object> Future<Boolean> hset(K key, K field, byte[] value) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] entryBytes = null;
if((entryBytes = JRedisSupport.getKeyBytes(field)) == null)
throw new IllegalArgumentException ("invalid field => ["+field+"]");
Future<Response> futureResponse = this.queueRequest(Command.HSET, keyBytes, entryBytes, value);
return new FutureBoolean(futureResponse);
}
public <K extends Object> Future<Boolean> hset(K key, K field, String stringValue) {
return hset (key, field, DefaultCodec.encode(stringValue));
}
public <K extends Object> Future<Long> hincrby(K key, K field, long increment) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] entryBytes = null;
if((entryBytes = JRedisSupport.getKeyBytes(field)) == null)
throw new IllegalArgumentException ("invalid field => ["+field+"]");
Future<Response> futureResponse = this.queueRequest(Command.HINCRBY, keyBytes, entryBytes, Convert.toBytes(increment));
return new FutureLong(futureResponse);
}
public <K extends Object> Future<Boolean> hset(K key, K field, Number numberValue) {
return hset (key, field, String.valueOf(numberValue).getBytes());
}
public <K extends Object, T extends Serializable>
Future<Boolean> hset(K key, K field, T object) {
return hset (key, field, DefaultCodec.encode(object));
}
public <K extends Object> Future<byte[]> hget(K key, K entry) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] entryBytes = null;
if((entryBytes = JRedisSupport.getKeyBytes(entry)) == null)
throw new IllegalArgumentException ("invalid field => ["+entry+"]");
Future<Response> futureResponse = this.queueRequest(Command.HGET, keyBytes, entryBytes);
return new FutureByteArray(futureResponse);
}
public <K extends Object> Future<Boolean> hexists(K key, K entry) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] entryBytes = null;
if((entryBytes = JRedisSupport.getKeyBytes(entry)) == null)
throw new IllegalArgumentException ("invalid field => ["+entry+"]");
Future<Response> futureResponse = this.queueRequest(Command.HEXISTS, keyBytes, entryBytes);
return new FutureBoolean(futureResponse);
}
public <K extends Object> Future<Boolean> hdel(K key, K entry) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] entryBytes = null;
if((entryBytes = JRedisSupport.getKeyBytes(entry)) == null)
throw new IllegalArgumentException ("invalid field => ["+entry+"]");
Future<Response> futureResponse = this.queueRequest(Command.HDEL, keyBytes, entryBytes);
return new FutureBoolean(futureResponse);
}
public <K extends Object> Future<Long> hlen(K key) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.HLEN, keyBytes);
return new FutureLong(futureResponse);
}
public <K extends Object> Future<List<byte[]>> hkeys(K key) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.HKEYS, keyBytes);
return new FutureKeyList (futureResponse);
}
@Override
public <K extends Object> Future<List<byte[]>> hvals(K key) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.HKEYS, keyBytes);
return new FutureByteArrayList (futureResponse);
}
@Override
public <K extends Object> Future<Map<byte[], byte[]>> hgetall(K key) {
byte[] keyBytes = null;
if((keyBytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.HGETALL, keyBytes);
return new FutureDataDictionary (futureResponse);
}
/* ------------------------------- commands returning int value --------- */
@Override
public <K extends Object> Future<Long> incr(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.INCR, keybytes);
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> incrby(K key, int delta) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.INCRBY, keybytes, Convert.toBytes(delta));
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> decr(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.DECR, keybytes);
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> decrby(K key, int delta) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.DECRBY, keybytes, Convert.toBytes(delta));
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> llen(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.LLEN, keybytes);
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> scard(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SCARD, keybytes);
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> zcard(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZCARD, keybytes);
return new FutureLong (futureResponse);
}
public <K extends Object> Future<byte[]> srandmember (K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SRANDMEMBER, keybytes);
return new FutureByteArray (futureResponse);
}
public <K extends Object> Future<byte[]> spop (K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SPOP, keybytes);
return new FutureByteArray (futureResponse);
}
/* ------------------------------- commands returning long value --------- */
@Override
public <K extends Object> Future<Long> dbsize() {
Future<Response> futureResponse = this.queueRequest(Command.DBSIZE);
return new FutureLong (futureResponse);
}
@Override
public <K extends Object> Future<Long> lastsave() {
Future<Response> futureResponse = this.queueRequest(Command.LASTSAVE);
return new FutureLong (futureResponse);
}
/* ------------------------------- commands returning byte[] --------- */
@Override
public <K extends Object> Future<byte[]> get(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.GET, keybytes);
return new FutureByteArray(futureResponse);
}
@Override
public <K extends Object> Future<byte[]> lindex(K key, long index) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.LINDEX, keybytes, Convert.toBytes(index));
return new FutureByteArray(futureResponse);
}
@Override
public <K extends Object> Future<byte[]> lpop(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.LPOP, keybytes);
return new FutureByteArray(futureResponse);
}
@Override
public <K extends Object> Future<byte[]> rpop(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.RPOP, keybytes);
return new FutureByteArray(futureResponse);
}
/* ------------------------------- commands returning String--------- */
@Override
public <K extends Object> Future<byte[]> randomkey() {
Future<Response> futureResponse = this.queueRequest(Command.RANDOMKEY);
return new FutureByteArray(futureResponse);
}
@Override
public <K extends Object> Future<RedisType> type(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
return new FutureRedisType(this.queueRequest(Command.TYPE, keybytes));
}
/* ------------------------------- commands returning Maps --------- */
@Override
public <K extends Object> Future<Map<String, String>> info() {
return new FutureInfo(this.queueRequest(Command.INFO));
}
@Override
public <K extends Object> Future<ObjectInfo> debug (K key) {
byte[] keybytes = JRedisSupport.getKeyBytes(key);
// if(key.length() == 0)
// throw new IllegalArgumentException ("invalid zero length key => ["+key+"]");
return new FutureObjectInfo (this.queueRequest(Command.DEBUG, "OBJECT".getBytes(), keybytes));
}
/* ------------------------------- commands returning Lists --------- */
@Override
public <K extends Object> Future<List<byte[]>> mget(String ... keys) {
if(null == keys || keys.length == 0) throw new IllegalArgumentException("no keys specified");
byte[] keydata = null;
byte[][] keybytes = new byte[keys.length][];
int i=0;
for(String k : keys) {
if((keydata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"] @ index: " + i);
keybytes[i++] = keydata;
}
return new FutureByteArrayList(this.queueRequest(Command.MGET, keybytes));
}
/* MSETs */
private FutureStatus mset(byte[][] mappings){
Future<Response> futureResponse = this.queueRequest(Command.MSET, mappings);
return new FutureStatus(futureResponse);
}
public <K extends Object> FutureStatus mset(Map<K, byte[]> keyValueMap){
// KeyCodec codec = DefaultKeyCodec.provider();
byte[][] mappings = new byte[keyValueMap.size()*2][];
int i = 0;
for (Entry<K, byte[]> e : keyValueMap.entrySet()){
mappings[i++] = JRedisSupport.getKeyBytes(e.getKey());
mappings[i++] = e.getValue();
}
return mset(mappings);
}
public <K extends Object> FutureStatus mset(KeyValueSet.ByteArrays<K> keyValueMap){
return mset(keyValueMap.getMappings());
}
public <K extends Object> FutureStatus mset(KeyValueSet.Strings<K> keyValueMap){
return mset(keyValueMap.getMappings());
}
public <K extends Object> FutureStatus mset(KeyValueSet.Numbers<K> keyValueMap){
return mset(keyValueMap.getMappings());
}
public <K extends Object, T extends Serializable> FutureStatus mset(KeyValueSet.Objects<K, T> keyValueMap){
return mset(keyValueMap.getMappings());
}
/* MSETNXs */
private Future<Boolean> msetnx(byte[][] mappings){
Future<Response> futureResponse = this.queueRequest(Command.MSETNX, mappings);
return new FutureBoolean(futureResponse);
}
public <K extends Object> Future<Boolean> msetnx(Map<K, byte[]> keyValueMap){
// KeyCodec codec = DefaultKeyCodec.provider();
byte[][] mappings = new byte[keyValueMap.size()*2][];
int i = 0;
for (Entry<K, byte[]> e : keyValueMap.entrySet()){
mappings[i++] = JRedisSupport.getKeyBytes(e.getKey());
mappings[i++] = e.getValue();
}
return msetnx(mappings);
}
public <K extends Object> Future<Boolean> msetnx(KeyValueSet.ByteArrays<K> keyValueMap){
return msetnx(keyValueMap.getMappings());
}
public <K extends Object> Future<Boolean> msetnx(KeyValueSet.Strings<K> keyValueMap){
return msetnx(keyValueMap.getMappings());
}
public <K extends Object> Future<Boolean> msetnx(KeyValueSet.Numbers<K> keyValueMap){
return msetnx(keyValueMap.getMappings());
}
public <K extends Object,T extends Serializable> Future<Boolean> msetnx(KeyValueSet.Objects<K, T> keyValueMap){
return msetnx(keyValueMap.getMappings());
}
@Override
public <K extends Object> Future<List<byte[]>> smembers(K key) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("null key.");
return new FutureByteArrayList(this.queueRequest(Command.SMEMBERS, keydata));
}
@Override
public <K extends Object> Future<List<byte[]>> keys() {
return this.keys("*");
}
@Override
public <K extends Object> Future<List<byte[]>> keys(K pattern) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(pattern)) == null)
throw new IllegalArgumentException ("null key.");
Future<Response> futureResponse = this.queueRequest(Command.KEYS, keydata);
return new FutureKeyList(futureResponse);
}
public <K extends Object> Future<Long> keystolist(String pattern, String listname) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(pattern)) == null)
throw new IllegalArgumentException ("null key.");
byte[] listnamedata = null;
if((listnamedata = JRedisSupport.getKeyBytes(listname)) == null)
throw new IllegalArgumentException ("null list name.");
return new FutureLong(this.queueRequest(Command.KEYSTOLIST, keydata, listnamedata));
}
@Override
public <K extends Object> Future<List<byte[]>> lrange(K key, long from, long to) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(from);
byte[] toBytes = Convert.toBytes(to);
return new FutureByteArrayList(this.queueRequest(Command.LRANGE, keybytes, fromBytes, toBytes));
}
@Override
public <K extends Object> Future<byte[]> substr(K key, long from, long to) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(from);
byte[] toBytes = Convert.toBytes(to);
return new FutureByteArray(this.queueRequest(Command.SUBSTR, keybytes, fromBytes, toBytes));
}
@Override
public <K extends Object> Future<List<byte[]>> zrange(K key, long from, long to) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(from);
byte[] toBytes = Convert.toBytes(to);
return new FutureByteArrayList(this.queueRequest(Command.ZRANGE, keybytes, fromBytes, toBytes));
}
@Override
public <K extends Object> Future<List<byte[]>> zrangebyscore(K key, double minScore, double maxScore) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] minScoreBytes = Convert.toBytes(minScore);
byte[] maxScoreBytes = Convert.toBytes(maxScore);
return new FutureByteArrayList(this.queueRequest(Command.ZRANGEBYSCORE, keybytes, minScoreBytes, maxScoreBytes));
}
@Override
public <K extends Object> Future<List<ZSetEntry>> zrangebyscoreSubset(K key, double minScore, double maxScore) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] minScoreBytes = Convert.toBytes(minScore);
byte[] maxScoreBytes = Convert.toBytes(maxScore);
return new FutureZSetList(this.queueRequest(Command.ZRANGEBYSCORE$OPTS, keybytes, minScoreBytes, maxScoreBytes, Command.Option.WITHSCORES.bytes));
}
@Override
public <K extends Object> Future<Long> zremrangebyscore(K key, double minScore, double maxScore) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] minScoreBytes = Convert.toBytes(minScore);
byte[] maxScoreBytes = Convert.toBytes(maxScore);
return new FutureLong(this.queueRequest(Command.ZREMRANGEBYSCORE, keybytes, minScoreBytes, maxScoreBytes));
}
@Override
public <K extends Object> Future<Long> zcount(K key, double minScore, double maxScore) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] minScoreBytes = Convert.toBytes(minScore);
byte[] maxScoreBytes = Convert.toBytes(maxScore);
return new FutureLong(this.queueRequest(Command.ZCOUNT, keybytes, minScoreBytes, maxScoreBytes));
}
@Override
public <K extends Object> Future<Long> zremrangebyrank(K key, long minRank, long maxRank) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] minScoreBytes = Convert.toBytes(minRank);
byte[] maxScoreBytes = Convert.toBytes(maxRank);
return new FutureLong(this.queueRequest(Command.ZREMRANGEBYRANK, keybytes, minScoreBytes, maxScoreBytes));
}
@Override
public <K extends Object> Future<List<byte[]>> zrevrange(K key, long from, long to) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(from);
byte[] toBytes = Convert.toBytes(to);
return new FutureByteArrayList(this.queueRequest(Command.ZREVRANGE, keybytes, fromBytes, toBytes));
}
@Override
public <K extends Object> Future<List<ZSetEntry>> zrangeSubset(K key, long from, long to) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(from);
byte[] toBytes = Convert.toBytes(to);
return new FutureZSetList(this.queueRequest(Command.ZRANGE$OPTS, keybytes, fromBytes, toBytes, Command.Option.WITHSCORES.bytes));
}
@Override
public <K extends Object> Future<List<ZSetEntry>> zrevrangeSubset(K key, long from, long to) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(from);
byte[] toBytes = Convert.toBytes(to);
return new FutureZSetList(this.queueRequest(Command.ZREVRANGE$OPTS, keybytes, fromBytes, toBytes, Command.Option.WITHSCORES.bytes));
}
@Override
public <K extends Object> Sort sort(final K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
final JRedisFutureSupport client = this;
Sort sortQuery = new SortSupport (keybytes) {
@Override
protected Future<List<byte[]>> execAsyncSort(byte[]... fullSortCmd) {
return new FutureByteArrayList(client.queueRequest(Command.SORT, fullSortCmd));
}
protected Future<List<byte[]>> execAsyncSortStore(byte[]... fullSortCmd) {
Future<Response> fResp = client.queueRequest(Command.SORT$STORE, fullSortCmd);
// new FutureLong(fResp); // hah?
return new FutureSortStoreResp(fResp);
}
protected List<byte[]> execSort(byte[]... fullSortCmd) {
throw new IllegalStateException("JRedisFuture does not support synchronous sort.");
}
protected List<byte[]> execSortStore(byte[]... fullSortCmd) {
throw new IllegalStateException("JRedisFuture does not support synchronous sort.");
}
};
return sortQuery;
}
/* ------------------------------- commands that don't get a response --------- */
@Override
public <K extends Object> FutureStatus quit() {
return new FutureStatus(this.queueRequest(Command.QUIT));
}
@Override
public FutureStatus flush() {
return new FutureStatus(this.queueRequest(Command.CONN_FLUSH));
}
/* ------------------------------- commands that don't get a response END --------- */
@Override
public <K extends Object> Future<List<byte[]>> sinter(K set1, K... sets) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(set1)) == null)
throw new IllegalArgumentException ("invalid key => ["+set1+"]");
byte[][] keybytes = new byte[1+sets.length][];
int i=0; keybytes[i++] = keydata;
for(K k : sets) {
if((keydata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"]");
keybytes[i++] = keydata;
}
return new FutureByteArrayList(this.queueRequest(Command.SINTER, keybytes));
}
@Override
public <K extends Object> Future<List<byte[]>> sunion(K set1, K... sets) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(set1)) == null)
throw new IllegalArgumentException ("invalid key => ["+set1+"]");
byte[][] keybytes = new byte[1+sets.length][];
int i=0; keybytes[i++] = keydata;
for(K k : sets) {
if((keydata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"]");
keybytes[i++] = keydata;
}
return new FutureByteArrayList(this.queueRequest(Command.SUNION, keybytes));
}
@Override
public <K extends Object> Future<List<byte[]>> sdiff(K set1, K... sets) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(set1)) == null)
throw new IllegalArgumentException ("invalid key => ["+set1+"]");
byte[][] keybytes = new byte[sets.length+1][];
int i=0; keybytes[i++] = keydata;
for(K k : sets) {
if((keydata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"]");
keybytes[i++] = keydata;
}
return new FutureByteArrayList(this.queueRequest(Command.SDIFF, keybytes));
}
@Override
public <K extends Object> FutureStatus sinterstore(K dest, K... sets) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(dest)) == null)
throw new IllegalArgumentException ("invalid key => ["+dest+"]");
byte[][] setbytes = new byte[1+sets.length][];
int i=0;
setbytes[i++] = keydata;
byte[] setdata =null;
for(K k : sets) {
if((setdata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"]");
setbytes[i++] = setdata;
}
return new FutureStatus(this.queueRequest(Command.SINTERSTORE, setbytes));
}
@Override
public <K extends Object> FutureStatus sunionstore(K dest, K... sets) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(dest)) == null)
throw new IllegalArgumentException ("invalid key => ["+dest+"]");
byte[][] setbytes = new byte[1+sets.length][];
int i=0;
setbytes[i++] = keydata;
byte[] setdata =null;
for(K k : sets) {
if((setdata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"]");
setbytes[i++] = setdata;
}
return new FutureStatus(this.queueRequest(Command.SUNIONSTORE, setbytes));
}
@Override
public <K extends Object> FutureStatus sdiffstore(K dest, K... sets) {
byte[] keydata = null;
if((keydata = JRedisSupport.getKeyBytes(dest)) == null)
throw new IllegalArgumentException ("invalid key => ["+dest+"]");
byte[][] setbytes = new byte[1+sets.length][];
int i=0;
setbytes[i++] = keydata;
byte[] setdata =null;
for(K k : sets) {
if((setdata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"]");
setbytes[i++] = setdata;
}
return new FutureStatus(this.queueRequest(Command.SDIFFSTORE, setbytes));
}
@Override
public <K extends Object> Future<Long> del(K ... keys) {
if(null == keys || keys.length == 0) throw new IllegalArgumentException("no keys specified");
byte[] keydata = null;
byte[][] keybytes = new byte[keys.length][];
int i=0;
for(K k : keys) {
if((keydata = JRedisSupport.getKeyBytes(k)) == null)
throw new IllegalArgumentException ("invalid key => ["+k+"] @ index: " + i);
keybytes[i++] = keydata;
}
Future<Response> futureResponse = this.queueRequest(Command.DEL, keybytes);
return new FutureLong(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> exists(K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.EXISTS, keybytes);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> FutureLong lpush(K key, byte[] value) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
if(value == null)
throw new IllegalArgumentException ("null value");
return new FutureLong(this.queueRequest(Command.LPUSH, keybytes, value));
}
@Override
public <K extends Object> FutureLong lpush(K key, String value) {
return lpush(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> FutureLong lpush(K key, Number value) {
return lpush(key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> FutureLong lpush (K key, T value)
{
return lpush(key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> lrem(K key, byte[] value, int count) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] countBytes = Convert.toBytes(count);
Future<Response> futureResponse = this.queueRequest(Command.LREM, keybytes, countBytes, value);
return new FutureLong(futureResponse);
}
@Override
public <K extends Object> Future<Long> lrem (K listKey, String value, int count){
return lrem (listKey, DefaultCodec.encode(value), count);
}
@Override
public <K extends Object> Future<Long> lrem (K listKey, Number numberValue, int count) {
return lrem (listKey, String.valueOf(numberValue).getBytes(), count);
}
@Override
public <K extends Object, T extends Serializable>
Future<Long> lrem (K listKey, T object, int count){
return lrem (listKey, DefaultCodec.encode(object), count);
}
@Override
public <K extends Object> FutureStatus lset(K key, long index, byte[] value) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] indexBytes = Convert.toBytes(index);
return new FutureStatus(this.queueRequest(Command.LSET, keybytes, indexBytes, value));
}
@Override
public <K extends Object> FutureStatus lset (K key, long index, String value) {
return lset (key, index, DefaultCodec.encode(value));
}
@Override
public <K extends Object> FutureStatus lset (K key, long index, Number numberValue){
return lset (key, index, String.valueOf(numberValue).getBytes());
}
@Override
public <K extends Object, T extends Serializable> FutureStatus lset (K key, long index, T object){
return lset (key, index, DefaultCodec.encode(object));
}
@Override
public <K extends Object> Future<Boolean> move(K key, int dbIndex) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.MOVE, keybytes, Convert.toBytes(dbIndex));
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> srem(K key, byte[] member) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.SREM, keybytes, member);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> srem (K key, String value) {
return srem (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> srem (K key, Number value) {
return srem (key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Boolean> srem (K key, T value)
{
return srem (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> zrem(K key, byte[] member) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZREM, keybytes, member);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> zrem (K key, String value) {
return zrem (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Boolean> zrem (K key, Number value) {
return zrem (key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Boolean> zrem (K key, T value)
{
return zrem (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Double> zscore(K key, byte[] member) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZSCORE, keybytes, member);
return new FutureDouble(futureResponse);
}
@Override
public <K extends Object> Future<Double> zscore (K key, String value) {
return zscore (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Double> zscore (K key, Number value) {
return zscore (key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Double> zscore (K key, T value)
{
return zscore (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> zrank(K key, byte[] member) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZRANK, keybytes, member);
return new FutureLong(futureResponse);
}
@Override
public <K extends Object> Future<Long> zrank (K key, String value) {
return zrank (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> zrank (K key, Number value) {
return zrank (key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Long> zrank (K key, T value)
{
return zrank (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> zrevrank(K key, byte[] member) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.ZREVRANK, keybytes, member);
return new FutureLong(futureResponse);
}
@Override
public <K extends Object> Future<Long> zrevrank (K key, String value) {
return zrevrank (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> Future<Long> zrevrank (K key, Number value) {
return zrevrank (key, String.valueOf(value).getBytes());
}
@Override
public <K extends Object, T extends Serializable> Future<Long> zrevrank (K key, T value)
{
return zrevrank (key, DefaultCodec.encode(value));
}
@Override
public <K extends Object> FutureStatus ltrim(K key, long keepFrom, long keepTo) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] fromBytes = Convert.toBytes(keepFrom);
byte[] toBytes = Convert.toBytes(keepTo);
return new FutureStatus(this.queueRequest(Command.LTRIM, keybytes, fromBytes, toBytes));
}
@Override
public <K extends Object> Future<Boolean> expire(K key, int ttlseconds) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
byte[] ttlbytes = Convert.toBytes(ttlseconds);
Future<Response> futureResponse = this.queueRequest(Command.EXPIRE, keybytes, ttlbytes);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Boolean> expireat(K key, long epochtime) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
long expiretime = TimeUnit.SECONDS.convert(epochtime, TimeUnit.MILLISECONDS);
byte[] expiretimeBytes = Convert.toBytes(expiretime);
Future<Response> futureResponse = this.queueRequest(Command.EXPIREAT, keybytes, expiretimeBytes);
return new FutureBoolean(futureResponse);
}
@Override
public <K extends Object> Future<Long> ttl (K key) {
byte[] keybytes = null;
if((keybytes = JRedisSupport.getKeyBytes(key)) == null)
throw new IllegalArgumentException ("invalid key => ["+key+"]");
Future<Response> futureResponse = this.queueRequest(Command.TTL, keybytes);
return new FutureLong(futureResponse);
}
// TODO: integrate using KeyCodec and a CodecManager at client spec and init time.
// TODO: (implied) ClientSpec (impls. ConnectionSpec)
// this isn't cooked yet -- lets think more about the implications...
//
// static final private Map<String, byte[]> keyByteCache = new ConcurrentHashMap<String, byte[]>();
public static final boolean CacheKeys = false;
public static class FutureResultBase {
final protected Future<Response> pendingRequest;
protected FutureResultBase(Future<Response> pendingRequest){ this.pendingRequest = pendingRequest;}
public boolean cancel (boolean mayInterruptIfRunning) {
return pendingRequest.cancel(mayInterruptIfRunning);
}
public boolean isCancelled () {
return pendingRequest.isCancelled();
}
public boolean isDone () {
return pendingRequest.isDone();
}
}
public static class FutureStatus extends FutureResultBase implements Future<ResponseStatus> {
protected FutureStatus (Future<Response> pendingRequest) { super(pendingRequest); }
public ResponseStatus get () throws InterruptedException, ExecutionException {
// StatusResponse statusResponse = (StatusResponse) pendingRequest.get(); // HANDLE VIRTUALS HERE
// return statusResponse.getStatus();
return pendingRequest.get().getStatus();
}
public ResponseStatus get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
// StatusResponse statusResponse = (StatusResponse) pendingRequest.get(timeout, unit);
// return statusResponse.getStatus();
return pendingRequest.get(timeout, unit).getStatus();
}
}
public static class FutureBoolean extends FutureResultBase implements Future<Boolean>{
protected FutureBoolean (Future<Response> pendingRequest) { super(pendingRequest); }
@SuppressWarnings("boxing")
public Boolean get () throws InterruptedException, ExecutionException {
ValueResponse valResp = (ValueResponse) pendingRequest.get();
return valResp.getBooleanValue();
}
@SuppressWarnings("boxing")
public Boolean get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse valResp = (ValueResponse) pendingRequest.get(timeout, unit);
return valResp.getBooleanValue();
}
}
public static class FutureBit extends FutureResultBase implements Future<Boolean> {
protected FutureBit (Future<Response> pendingRequest) { super(pendingRequest); }
@SuppressWarnings("boxing")
public Boolean get () throws InterruptedException, ExecutionException {
ValueResponse valResp = (ValueResponse) pendingRequest.get();
return valResp.getLongValue() == 1;
}
@SuppressWarnings("boxing")
public Boolean get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse valResp = (ValueResponse) pendingRequest.get(timeout, unit);
return valResp.getLongValue() == 1;
}
}
public static class FutureString extends FutureResultBase implements Future<String>{
protected FutureString (Future<Response> pendingRequest) { super(pendingRequest); }
public String get () throws InterruptedException, ExecutionException {
ValueResponse valResp = (ValueResponse) pendingRequest.get();
return valResp.getStringValue();
}
public String get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse valResp = (ValueResponse) pendingRequest.get(timeout, unit);
return valResp.getStringValue();
}
}
public static class FutureRedisType extends FutureResultBase implements Future<RedisType>{
protected FutureRedisType (Future<Response> pendingRequest) { super(pendingRequest); }
private final RedisType getRedisType(ValueResponse resp){
String stringValue = resp.getStringValue();
return RedisType.valueOf(stringValue);
}
public RedisType get () throws InterruptedException, ExecutionException {
ValueResponse valResp = (ValueResponse) pendingRequest.get();
return getRedisType(valResp);
}
public RedisType get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse valResp = (ValueResponse) pendingRequest.get(timeout, unit);
return getRedisType(valResp);
}
}
public static class FutureLong extends FutureResultBase implements Future<Long>{
protected FutureLong (Future<Response> pendingRequest) { super(pendingRequest); }
@SuppressWarnings("boxing")
public Long get () throws InterruptedException, ExecutionException {
ValueResponse valResp = (ValueResponse) pendingRequest.get();
return valResp.getLongValue();
}
@SuppressWarnings("boxing")
public Long get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse valResp = (ValueResponse) pendingRequest.get(timeout, unit);
return valResp.getLongValue();
}
}
public static class FutureDouble extends FutureResultBase implements Future<Double>{
protected FutureDouble (Future<Response> pendingRequest) { super(pendingRequest); }
@SuppressWarnings("boxing")
public Double get () throws InterruptedException, ExecutionException {
BulkResponse bulkResp = (BulkResponse) pendingRequest.get();
if(bulkResp.getBulkData() != null)
return Convert.toDouble(bulkResp.getBulkData());
return null;
}
@SuppressWarnings("boxing")
public Double get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
BulkResponse bulkResp = (BulkResponse) pendingRequest.get(timeout, unit);
if(bulkResp.getBulkData() != null)
return Convert.toDouble(bulkResp.getBulkData());
return null;
}
}
public static class FutureByteArray extends FutureResultBase implements Future<byte[]>{
protected FutureByteArray (Future<Response> pendingRequest) { super(pendingRequest); }
public byte[] get () throws InterruptedException, ExecutionException {
BulkResponse resp = (BulkResponse) pendingRequest.get();
return resp.getBulkData();
}
public byte[] get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
BulkResponse resp = (BulkResponse) pendingRequest.get(timeout, unit);
return resp.getBulkData();
}
}
public static class FutureSortStoreResp extends FutureResultBase implements Future<List<byte[]>>{
protected FutureSortStoreResp (Future<Response> pendingRequest) { super(pendingRequest); }
public List<byte[]> get () throws InterruptedException, ExecutionException {
ValueResponse resp = (ValueResponse) pendingRequest.get();
return packValueResult(resp.getLongValue());
}
public List<byte[]> get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse resp = (ValueResponse) pendingRequest.get(timeout, unit);
return packValueResult(resp.getLongValue());
}
private static List<byte[]> packValueResult(long number) {
List<byte[]> list = new ArrayList<byte[]>(1);
list.add(Convert.toBytes(number));
return list;
}
}
public static class FutureByteArrayList extends FutureResultBase implements Future<List<byte[]>>{
protected FutureByteArrayList (Future<Response> pendingRequest) { super(pendingRequest); }
public List<byte[]> get () throws InterruptedException, ExecutionException {
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get();
return resp.getMultiBulkData();
}
public List<byte[]> get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get(timeout, unit);
return resp.getMultiBulkData();
}
}
public static class FutureDataDictionary extends FutureResultBase implements Future<Map<byte[], byte[]>>{
protected FutureDataDictionary (Future<Response> pendingRequest) { super(pendingRequest); }
public Map<byte[], byte[]> get () throws InterruptedException, ExecutionException {
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get();
return convert(resp.getMultiBulkData());
}
public Map<byte[], byte[]> get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get(timeout, unit);
return convert(resp.getMultiBulkData());
}
private static final Map<byte[], byte[]> convert (List<byte[]> bulkdata) {
Map<byte[], byte[]> map = null;
if(null != bulkdata) {
map = new HashMap<byte[], byte[]>(bulkdata.size()/2);
for(int i=0; i<bulkdata.size(); i+=2){
// map.put(DefaultCodec.toStr(bulkdata.get(i)), bulkdata.get(i+1));
map.put(bulkdata.get(i), bulkdata.get(i+1));
}
}
return map;
}
}
public static class FutureKeyList extends FutureResultBase implements Future<List<byte[]>>{
protected FutureKeyList (Future<Response> pendingRequest) { super(pendingRequest); }
// private List<String> getResultList (BulkResponse resp) {
// StringTokenizer tokenizer = new StringTokenizer(new String(resp.getBulkData()), " ");
// List<String> list = new ArrayList <String>(12);
// while (tokenizer.hasMoreTokens()){
// list.add(tokenizer.nextToken());
// }
// return list;
// }
public List<byte[]> get () throws InterruptedException, ExecutionException {
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get();
List<byte[]> multibulkdata = resp.getMultiBulkData();
// List<String> list = null;
// if(null != multibulkdata)
// list = DefaultCodec.toStr(multibulkdata);
return multibulkdata;
}
public List<byte[]> get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get(timeout, unit);
List<byte[]> multibulkdata = resp.getMultiBulkData();
// List<String> list = null;
// if(null != multibulkdata)
// list = DefaultCodec.toStr(multibulkdata);
return multibulkdata;
}
}
public static class FutureInfo extends FutureResultBase implements Future<Map<String, String>>{
protected FutureInfo (Future<Response> pendingRequest) { super(pendingRequest); }
private Map<String, String> getResultMap (BulkResponse resp) {
StringTokenizer tokenizer = new StringTokenizer(new String(resp.getBulkData()), "\r\n");
Map<String, String> infomap = new HashMap<String, String>(48);
while (tokenizer.hasMoreTokens()){
final String info = tokenizer.nextToken();
// ignore comments "# heading"
if(info.startsWith("#"))
continue;
int c = info.indexOf(':');
String _key =info.substring(0, c);
String _value = info.substring(c+1);
infomap.put(_key, _value);
}
return infomap;
}
public Map<String, String> get () throws InterruptedException, ExecutionException {
BulkResponse resp = (BulkResponse) pendingRequest.get();
return getResultMap(resp);
}
public Map<String, String> get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
BulkResponse resp = (BulkResponse) pendingRequest.get(timeout, unit);
return getResultMap(resp);
}
}
public static class FutureObjectInfo extends FutureResultBase implements Future<ObjectInfo>{
protected FutureObjectInfo (Future<Response> pendingRequest) { super(pendingRequest); }
private final ObjectInfo getObjectInfo(ValueResponse resp){
String stringValue = resp.getStringValue();
return ObjectInfo.valueOf(stringValue);
}
public ObjectInfo get () throws InterruptedException, ExecutionException {
ValueResponse valResp = (ValueResponse) pendingRequest.get();
return getObjectInfo(valResp);
}
public ObjectInfo get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
ValueResponse valResp = (ValueResponse) pendingRequest.get(timeout, unit);
return getObjectInfo(valResp);
}
}
public static class FutureZSetList extends FutureResultBase implements Future<List<ZSetEntry>>{
protected FutureZSetList (Future<Response> pendingRequest) { super(pendingRequest); }
public List<ZSetEntry> get () throws InterruptedException, ExecutionException {
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get();
return convert(resp.getMultiBulkData());
}
public List<ZSetEntry> get (long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
MultiBulkResponse resp = (MultiBulkResponse) pendingRequest.get(timeout, unit);
return convert(resp.getMultiBulkData());
}
private static final List<ZSetEntry> convert (List<byte[]> mbulkdata) {
List<ZSetEntry> zset = null;
if(mbulkdata.size() > 0){
zset = new ArrayList<ZSetEntry>(mbulkdata.size()/2);
for(int i=0; i<mbulkdata.size(); i+=2){
zset.add(new ZSetEntryImpl(mbulkdata.get(i), mbulkdata.get(i+1)));
}
}
return zset;
}
}
// ------------------------------------------------------------------------
// Diagnostics commands
// ------------------------------------------------------------------------
/**
* @Redis ECHO
* @param msg
* @return
*/
public <K extends Object> Future<byte[]> echo (byte[] msg) {
if(msg == null)
throw new IllegalArgumentException ("invalid value for echo => [null]");
Future<Response> futureResponse = this.queueRequest(Command.ECHO, msg);
return new FutureByteArray(futureResponse);
}
public <K extends Object> Future<byte[]> echo (String msg) {
return echo(DefaultCodec.encode(msg));
}
public <K extends Object> Future<byte[]> echo (Number msg) {
return echo(String.valueOf(msg).getBytes());
}
public <K extends Object, T extends Serializable>
Future<byte[]> echo (T msg) {
return echo (DefaultCodec.encode(msg));
}
}