/*

+----------------------------------------------------------------------+

| Copyright (c) 1997-2009 The PHP Group |

+----------------------------------------------------------------------+

| This source file is subject to version 3.01 of the PHP license, |

| that is bundled with this package in the file LICENSE, and is |

| available through the world-wide-web at the following url: |

| http://www.php.net/license/3_01.txt |

| If you did not receive a copy of the PHP license and are unable to |

| obtain it through the world-wide-web, please send a note to |

| license@php.net so we can mail you a copy immediately. |

+----------------------------------------------------------------------+

| Author: Michael Grunder <michael.grunder@gmail.com> |

| Maintainer: Nicolas Favre-Felix <n.favre-felix@owlient.eu> |

+----------------------------------------------------------------------+

*/

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include "common.h"

#include "php_redis.h"

#include "ext/standard/info.h"

#include "crc16.h"

#include "redis_cluster.h"

#include "redis_commands.h"

#include <ext/spl/spl_exceptions.h>

#include <zend_exceptions.h>

#include "library.h"

#include <php_variables.h>

#include <SAPI.h>

zend_class_entry *redis_cluster_ce;

/* Exception handler */

zend_class_entry *redis_cluster_exception_ce;

#if PHP_VERSION_ID < 80000

#include "redis_cluster_legacy_arginfo.h"

#else

#include "zend_attributes.h"

#include "redis_cluster_arginfo.h"

#endif

PHP_MINIT_FUNCTION(redis_cluster)

{

redis_cluster_ce = register_class_RedisCluster();

redis_cluster_ce->create_object = create_cluster_context;

redis_cluster_exception_ce = register_class_RedisClusterException(spl_ce_RuntimeException);

return SUCCESS;

}

/* Handlers for RedisCluster */

zend_object_handlers RedisCluster_handlers;

/* Our context seeds will be a hash table with RedisSock* pointers */

static void ht_free_seed(zval *data) {

RedisSock *redis_sock = *(RedisSock**)data;

if (redis_sock) redis_free_socket(redis_sock);

}

/* Free redisClusterNode objects we've stored */

static void ht_free_node(zval *data) {

redisClusterNode *node = *(redisClusterNode**)data;

cluster_free_node(node);

}

/* Create redisCluster context */

zend_object * create_cluster_context(zend_class_entry *class_type) {

redisCluster *cluster;

// Allocate our actual struct

cluster = ecalloc(1, sizeof(redisCluster) + zend_object_properties_size(class_type));

// We're not currently subscribed anywhere

cluster->subscribed_slot = -1;

// Allocate our RedisSock we'll use to store prefix/serialization flags

cluster->flags = ecalloc(1, sizeof(RedisSock));

// Allocate our hash table for seeds

ALLOC_HASHTABLE(cluster->seeds);

zend_hash_init(cluster->seeds, 0, NULL, ht_free_seed, 0);

// Allocate our hash table for connected Redis objects

ALLOC_HASHTABLE(cluster->nodes);

zend_hash_init(cluster->nodes, 0, NULL, ht_free_node, 0);

// Initialize it

zend_object_std_init(&cluster->std, class_type);

object_properties_init(&cluster->std, class_type);

memcpy(&RedisCluster_handlers, zend_get_std_object_handlers(), sizeof(RedisCluster_handlers));

RedisCluster_handlers.offset = XtOffsetOf(redisCluster, std);

RedisCluster_handlers.free_obj = free_cluster_context;

cluster->std.handlers = &RedisCluster_handlers;

return &cluster->std;

}

/* Free redisCluster context */

void free_cluster_context(zend_object *object) {

redisCluster *cluster = PHPREDIS_GET_OBJECT(redisCluster, object);

cluster_free(cluster, 0);

zend_object_std_dtor(&cluster->std);

}

/* Take user provided seeds and return unique and valid ones */

/* Attempt to connect to a Redis cluster provided seeds and timeout options */

static void redis_cluster_init(redisCluster *c, HashTable *ht_seeds, double timeout,

double read_timeout, int persistent, zend_string *user,

zend_string *pass, zval *context)

{

zend_string *hash = NULL, **seeds;

redisCachedCluster *cc;

uint32_t nseeds;

char *err;

/* Validate our arguments and get a sanitized seed array */

seeds = cluster_validate_args(timeout, read_timeout, ht_seeds, &nseeds, &err);

if (seeds == NULL) {

CLUSTER_THROW_EXCEPTION(err, 0);

return;

}

if (user && ZSTR_LEN(user))

c->flags->user = zend_string_copy(user);

if (pass && ZSTR_LEN(pass))

c->flags->pass = zend_string_copy(pass);

if (context) {

redis_sock_set_stream_context(c->flags, context);

}

c->flags->timeout = timeout;

c->flags->read_timeout = read_timeout;

c->flags->persistent = persistent;

c->waitms = timeout * 1000L;

/* Attempt to load slots from cache if caching is enabled */

if (CLUSTER_CACHING_ENABLED()) {

/* Exit early if we can load from cache */

hash = cluster_hash_seeds(seeds, nseeds);

if ((cc = cluster_cache_load(hash))) {

cluster_init_cache(c, cc);

goto cleanup;

}

}

/* Initialize seeds and attempt to map keyspace */

cluster_init_seeds(c, seeds, nseeds);

if (cluster_map_keyspace(c) == SUCCESS && hash)

cluster_cache_store(hash, c->nodes);

cleanup:

if (hash) zend_string_release(hash);

free_seed_array(seeds, nseeds);

}

/* Attempt to load a named cluster configured in php.ini */

void redis_cluster_load(redisCluster *c, char *name, int name_len) {

zval z_seeds, z_tmp, *z_value;

zend_string *user = NULL, *pass = NULL;

double timeout = 0, read_timeout = 0;

int persistent = 0;

char *iptr;

HashTable *ht_seeds = NULL;

/* Seeds */

array_init(&z_seeds);

if ((iptr = INI_STR("redis.clusters.seeds")) != NULL) {

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_seeds);

}

if ((z_value = zend_hash_str_find(Z_ARRVAL(z_seeds), name, name_len)) != NULL) {

ht_seeds = Z_ARRVAL_P(z_value);

} else {

zval_dtor(&z_seeds);

CLUSTER_THROW_EXCEPTION("Couldn't find seeds for cluster", 0);

return;

}

/* Connection timeout */

if ((iptr = INI_STR("redis.clusters.timeout")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_double(Z_ARRVAL(z_tmp), name, name_len, &timeout);

zval_dtor(&z_tmp);

}

/* Read timeout */

if ((iptr = INI_STR("redis.clusters.read_timeout")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_double(Z_ARRVAL(z_tmp), name, name_len, &read_timeout);

zval_dtor(&z_tmp);

}

/* Persistent connections */

if ((iptr = INI_STR("redis.clusters.persistent")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_bool(Z_ARRVAL(z_tmp), name, name_len, &persistent);

zval_dtor(&z_tmp);

}

if ((iptr = INI_STR("redis.clusters.auth"))) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_auth(Z_ARRVAL(z_tmp), name, name_len, &user, &pass);

zval_dtor(&z_tmp);

}

/* Attempt to create/connect to the cluster */

redis_cluster_init(c, ht_seeds, timeout, read_timeout, persistent, user, pass, NULL);

/* Clean up */

zval_dtor(&z_seeds);

if (user) zend_string_release(user);

if (pass) zend_string_release(pass);

}

/*

* PHP Methods

*/

/* Create a RedisCluster Object */

PHP_METHOD(RedisCluster, __construct) {

zval *object, *z_seeds = NULL, *z_auth = NULL, *context = NULL;

zend_string *user = NULL, *pass = NULL;

double timeout = 0.0, read_timeout = 0.0;

size_t name_len;

zend_bool persistent = 0;

redisCluster *c = GET_CONTEXT();

char *name;

// Parse arguments

if (zend_parse_method_parameters(ZEND_NUM_ARGS(), getThis(),

"Os!|addbza!", &object, redis_cluster_ce, &name,

&name_len, &z_seeds, &timeout, &read_timeout,

&persistent, &z_auth, &context) == FAILURE)

{

RETURN_FALSE;

}

/* If we've got a string try to load from INI */

if (ZEND_NUM_ARGS() < 2) {

if (name_len == 0) { // Require a name

CLUSTER_THROW_EXCEPTION("You must specify a name or pass seeds!", 0);

}

redis_cluster_load(c, name, name_len);

return;

}

/* The normal case, loading from arguments */

redis_extract_auth_info(z_auth, &user, &pass);

redis_cluster_init(c, Z_ARRVAL_P(z_seeds), timeout, read_timeout,

persistent, user, pass, context);

if (user) zend_string_release(user);

if (pass) zend_string_release(pass);

}

/*

* RedisCluster method implementation

*/

/* {{{ proto bool RedisCluster::close() */

PHP_METHOD(RedisCluster, close) {

cluster_disconnect(GET_CONTEXT(), 1);

RETURN_TRUE;

}

/* {{{ proto string RedisCluster::get(string key) */

PHP_METHOD(RedisCluster, get) {

CLUSTER_PROCESS_KW_CMD("GET", redis_key_cmd, cluster_bulk_resp, 1);

}

/* }}} */

/* {{{ proto bool RedisCluster::set(string key, string value) */

PHP_METHOD(RedisCluster, set) {

CLUSTER_PROCESS_CMD(set, cluster_set_resp, 0);

}

/* }}} */

/* Generic handler for MGET/MSET/MSETNX */

static int

distcmd_resp_handler(INTERNAL_FUNCTION_PARAMETERS, redisCluster *c, short slot,

clusterMultiCmd *mc, zval *z_ret, int last, cluster_cb cb)

{

clusterMultiCtx *ctx;

// Finalize multi command

cluster_multi_fini(mc);

// Spin up multi context

ctx = emalloc(sizeof(clusterMultiCtx));

ctx->z_multi = z_ret;

ctx->count = mc->argc;

ctx->last = last;

// Attempt to send the command

if (cluster_send_command(c,slot,mc->cmd.c,mc->cmd.len) < 0 || c->err != NULL) {

efree(ctx);

return -1;

}

if (CLUSTER_IS_ATOMIC(c)) {

// Process response now

cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, c, (void*)ctx);

} else {

CLUSTER_ENQUEUE_RESPONSE(c, slot, cb, ctx);

}

// Clear out our command but retain allocated memory

CLUSTER_MULTI_CLEAR(mc);

return 0;

}

/* Container struct for a key/value pair pulled from an array */

typedef struct clusterKeyValHT {

char kbuf[22];

char *key;

size_t key_len;

int key_free;

short slot;

char *val;

size_t val_len;

int val_free;

} clusterKeyValHT;

/* Helper to pull a key/value pair from a HashTable */

static int get_key_val_ht(redisCluster *c, HashTable *ht, HashPosition *ptr,

clusterKeyValHT *kv)

{

zval *z_val;

zend_ulong idx;

// Grab the key, convert it to a string using provided kbuf buffer if it's

// a LONG style key

zend_string *zkey;

switch (zend_hash_get_current_key_ex(ht, &zkey, &idx, ptr)) {

case HASH_KEY_IS_STRING:

kv->key_len = ZSTR_LEN(zkey);

kv->key = ZSTR_VAL(zkey);

break;

case HASH_KEY_IS_LONG:

kv->key_len = snprintf(kv->kbuf,sizeof(kv->kbuf),"%ld",(long)idx);

kv->key = kv->kbuf;

break;

default:

CLUSTER_THROW_EXCEPTION("Internal Zend HashTable error", 0);

return -1;

}

// Prefix our key if we need to, set the slot

kv->key_free = redis_key_prefix(c->flags, &(kv->key), &(kv->key_len));

kv->slot = cluster_hash_key(kv->key, kv->key_len);

// Now grab our value

if ((z_val = zend_hash_get_current_data_ex(ht, ptr)) == NULL) {

CLUSTER_THROW_EXCEPTION("Internal Zend HashTable error", 0);

return -1;

}

// Serialize our value if required

kv->val_free = redis_pack(c->flags,z_val,&(kv->val),&(kv->val_len));

// Success

return 0;

}

/* Helper to pull, prefix, and hash a key from a HashTable value */

static int get_key_ht(redisCluster *c, HashTable *ht, HashPosition *ptr,

clusterKeyValHT *kv)

{

zval *z_key;

if ((z_key = zend_hash_get_current_data_ex(ht, ptr)) == NULL) {

// Shouldn't happen, but check anyway

CLUSTER_THROW_EXCEPTION("Internal Zend HashTable error", 0);

return -1;

}

// Always want to work with strings

convert_to_string(z_key);

kv->key = Z_STRVAL_P(z_key);

kv->key_len = Z_STRLEN_P(z_key);

kv->key_free = redis_key_prefix(c->flags, &(kv->key), &(kv->key_len));

// Hash our key

kv->slot = cluster_hash_key(kv->key, kv->key_len);

// Success

return 0;

}

/* Turn variable arguments into a HashTable for processing */

static HashTable *method_args_to_ht(zval *z_args, int argc) {

HashTable *ht_ret;

int i;

/* Allocate our hash table */

ALLOC_HASHTABLE(ht_ret);

zend_hash_init(ht_ret, argc, NULL, NULL, 0);

/* Populate our return hash table with our arguments */

for (i = 0; i < argc; i++) {

zend_hash_next_index_insert(ht_ret, &z_args[i]);

}

/* Return our hash table */

return ht_ret;

}

/* Convenience handler for commands that take multiple keys such as

* MGET, DEL, and UNLINK */

static int cluster_mkey_cmd(INTERNAL_FUNCTION_PARAMETERS, char *kw, int kw_len,

zval *z_ret, cluster_cb cb)

{

redisCluster *c = GET_CONTEXT();

clusterMultiCmd mc = {0};

clusterKeyValHT kv;

zval *z_args;

HashTable *ht_arr;

HashPosition ptr;

int i = 1, argc = ZEND_NUM_ARGS(), ht_free = 0;

short slot;

/* If we don't have any arguments we're invalid */

if (!argc) return -1;

/* Extract our arguments into an array */

z_args = ecalloc(argc, sizeof(zval));

if (zend_get_parameters_array(ht, argc, z_args) == FAILURE) {

efree(z_args);

return -1;

}

/* Determine if we're working with a single array or variadic args */

if (argc == 1 && Z_TYPE(z_args[0]) == IS_ARRAY) {

ht_arr = Z_ARRVAL(z_args[0]);

argc = zend_hash_num_elements(ht_arr);

if (!argc) {

efree(z_args);

return -1;

}

} else {

ht_arr = method_args_to_ht(z_args, argc);

ht_free = 1;

}

/* MGET is readonly, DEL is not */

c->readonly = kw_len == 4 && CLUSTER_IS_ATOMIC(c);

// Initialize our "multi" command handler with command/len

CLUSTER_MULTI_INIT(mc, kw, kw_len);

// Process the first key outside of our loop, so we don't have to check if

// it's the first iteration every time, needlessly

zend_hash_internal_pointer_reset_ex(ht_arr, &ptr);

if (get_key_ht(c, ht_arr, &ptr, &kv) < 0) {

efree(z_args);

return -1;

}

// Process our key and add it to the command

cluster_multi_add(&mc, kv.key, kv.key_len);

// Free key if we prefixed

if (kv.key_free) efree(kv.key);

// Move to the next key

zend_hash_move_forward_ex(ht_arr, &ptr);

// Iterate over keys 2...N

slot = kv.slot;

while (zend_hash_has_more_elements_ex(ht_arr, &ptr) ==SUCCESS) {

if (get_key_ht(c, ht_arr, &ptr, &kv) < 0) {

cluster_multi_free(&mc);

if (ht_free) {

zend_hash_destroy(ht_arr);

efree(ht_arr);

}

efree(z_args);

return -1;

}

// If the slots have changed, kick off the keys we've aggregated

if (slot != kv.slot) {

// Process this batch of MGET keys

if (distcmd_resp_handler(INTERNAL_FUNCTION_PARAM_PASSTHRU, c, slot,

&mc, z_ret, i == argc, cb) < 0)

{

cluster_multi_free(&mc);

if (ht_free) {

zend_hash_destroy(ht_arr);

efree(ht_arr);

}

efree(z_args);

return -1;

}

}

// Add this key to the command

cluster_multi_add(&mc, kv.key, kv.key_len);

// Free key if we prefixed

if (kv.key_free) efree(kv.key);

// Update the last slot we encountered, and the key we're on

slot = kv.slot;

i++;

zend_hash_move_forward_ex(ht_arr, &ptr);

}

efree(z_args);

// If we've got straggler(s) process them

if (mc.argc > 0) {

if (distcmd_resp_handler(INTERNAL_FUNCTION_PARAM_PASSTHRU, c, slot,

&mc, z_ret, 1, cb) < 0)

{

cluster_multi_free(&mc);

if (ht_free) {

zend_hash_destroy(ht_arr);

efree(ht_arr);

}

return -1;

}

}

// Free our command

cluster_multi_free(&mc);

/* Clean up our hash table if we constructed it from variadic args */

if (ht_free) {

zend_hash_destroy(ht_arr);

efree(ht_arr);

}

/* Return our object if we're in MULTI mode */

if (!CLUSTER_IS_ATOMIC(c))

RETVAL_ZVAL(getThis(), 1, 0);

// Success

return 0;

}

/* Handler for both MSET and MSETNX */

static int cluster_mset_cmd(INTERNAL_FUNCTION_PARAMETERS, char *kw, int kw_len,

zval *z_ret, cluster_cb cb)

{

redisCluster *c = GET_CONTEXT();

clusterKeyValHT kv;

clusterMultiCmd mc = {0};

zval *z_arr;

HashTable *ht_arr;

HashPosition ptr;

int i = 1, argc;

short slot;

// Parse our arguments

if (zend_parse_parameters(ZEND_NUM_ARGS(), "a", &z_arr) == FAILURE) {

return -1;

}

// No reason to send zero args

ht_arr = Z_ARRVAL_P(z_arr);

if ((argc = zend_hash_num_elements(ht_arr)) == 0) {

return -1;

}

/* This is a write command */

c->readonly = 0;

// Set up our multi command handler

CLUSTER_MULTI_INIT(mc, kw, kw_len);

// Process the first key/value pair outside of our loop

zend_hash_internal_pointer_reset_ex(ht_arr, &ptr);

if (get_key_val_ht(c, ht_arr, &ptr, &kv) ==-1) return -1;

zend_hash_move_forward_ex(ht_arr, &ptr);

// Add this to our multi cmd, set slot, free key if we prefixed

cluster_multi_add(&mc, kv.key, kv.key_len);

cluster_multi_add(&mc, kv.val, kv.val_len);

if (kv.key_free) efree(kv.key);

if (kv.val_free) efree(kv.val);

// While we've got more keys to set

slot = kv.slot;

while (zend_hash_has_more_elements_ex(ht_arr, &ptr) ==SUCCESS) {

// Pull the next key/value pair

if (get_key_val_ht(c, ht_arr, &ptr, &kv) ==-1) {

return -1;

}

// If the slots have changed, process responses

if (slot != kv.slot) {

if (distcmd_resp_handler(INTERNAL_FUNCTION_PARAM_PASSTHRU, c,

slot, &mc, z_ret, i == argc, cb) < 0)

{

cluster_multi_free(&mc);

return -1;

}

}

// Add this key and value to our command

cluster_multi_add(&mc, kv.key, kv.key_len);

cluster_multi_add(&mc, kv.val, kv.val_len);

// Free our key and value if we need to

if (kv.key_free) efree(kv.key);

if (kv.val_free) efree(kv.val);

// Update our slot, increment position

slot = kv.slot;

i++;

// Move on

zend_hash_move_forward_ex(ht_arr, &ptr);

}

// If we've got stragglers, process them too

if (mc.argc > 0) {

if (distcmd_resp_handler(INTERNAL_FUNCTION_PARAM_PASSTHRU, c, slot, &mc,

z_ret, 1, cb) < 0)

{

cluster_multi_free(&mc);

return -1;

}

}

// Free our command

cluster_multi_free(&mc);

/* Return our object if we're in MULTI mode */

if (!CLUSTER_IS_ATOMIC(c))

RETVAL_ZVAL(getThis(), 1, 0);

// Success

return 0;

}

/* Generic passthru for DEL and UNLINK which act identically */

static void cluster_generic_delete(INTERNAL_FUNCTION_PARAMETERS,

char *kw, int kw_len)

{

zval *z_ret = emalloc(sizeof(*z_ret));

// Initialize a LONG value to zero for our return

ZVAL_LONG(z_ret, 0);

// Parse args, process

if (cluster_mkey_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, kw, kw_len, z_ret,

cluster_del_resp) < 0)

{

efree(z_ret);

RETURN_FALSE;

}

}

/* {{{ proto array RedisCluster::del(string key1, string key2, ... keyN) */

PHP_METHOD(RedisCluster, del) {

cluster_generic_delete(INTERNAL_FUNCTION_PARAM_PASSTHRU, "DEL", sizeof("DEL") - 1);

}

/* {{{ proto array RedisCluster::unlink(string key1, string key2, ... keyN) */

PHP_METHOD(RedisCluster, unlink) {

cluster_generic_delete(INTERNAL_FUNCTION_PARAM_PASSTHRU, "UNLINK", sizeof("UNLINK") - 1);

}

/* {{{ proto array RedisCluster::mget(array keys) */

PHP_METHOD(RedisCluster, mget) {

zval *z_ret = emalloc(sizeof(*z_ret));

array_init(z_ret);

// Parse args, process

if (cluster_mkey_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "MGET",

sizeof("MGET")-1, z_ret, cluster_mbulk_mget_resp) < 0)

{

zval_dtor(z_ret);

efree(z_ret);

RETURN_FALSE;

}

}

/* {{{ proto bool RedisCluster::mset(array keyvalues) */

PHP_METHOD(RedisCluster, mset) {

zval *z_ret = emalloc(sizeof(*z_ret));

ZVAL_TRUE(z_ret);

// Parse args and process. If we get a failure, free zval and return FALSE.

if (cluster_mset_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "MSET",

sizeof("MSET")-1, z_ret, cluster_mset_resp) ==-1)

{

efree(z_ret);

RETURN_FALSE;

}

}

/* {{{ proto array RedisCluster::msetnx(array keyvalues) */

PHP_METHOD(RedisCluster, msetnx) {

zval *z_ret = emalloc(sizeof(*z_ret));

array_init(z_ret);

// Parse args and process. If we get a failure, free mem and return FALSE

if (cluster_mset_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "MSETNX",

sizeof("MSETNX")-1, z_ret, cluster_msetnx_resp) ==-1)

{

zval_dtor(z_ret);

efree(z_ret);

RETURN_FALSE;

}

}

/* }}} */

/* {{{ proto bool RedisCluster::setex(string key, string value, int expiry) */

PHP_METHOD(RedisCluster, setex) {

CLUSTER_PROCESS_KW_CMD("SETEX", redis_key_long_val_cmd, cluster_bool_resp, 0);

}

/* }}} */

/* {{{ proto bool RedisCluster::psetex(string key, string value, int expiry) */

PHP_METHOD(RedisCluster, psetex) {

CLUSTER_PROCESS_KW_CMD("PSETEX", redis_key_long_val_cmd, cluster_bool_resp, 0);

}

/* }}} */

/* {{{ proto bool RedisCluster::setnx(string key, string value) */

PHP_METHOD(RedisCluster, setnx) {

CLUSTER_PROCESS_KW_CMD("SETNX", redis_kv_cmd, cluster_1_resp, 0);

}

/* }}} */

/* {{{ proto string RedisCluster::getSet(string key, string value) */

PHP_METHOD(RedisCluster, getset) {

CLUSTER_PROCESS_KW_CMD("GETSET", redis_kv_cmd, cluster_bulk_resp, 0);

}

/* }}} */

/* {{{ proto int RedisCluster::exists(string key) */

PHP_METHOD(RedisCluster, exists) {

CLUSTER_PROCESS_CMD(exists, cluster_long_resp, 1);

}

/* }}} */

/* {{{ proto array Redis::keys(string pattern) */

PHP_METHOD(RedisCluster, keys) {

redisCluster *c = GET_CONTEXT();

redisClusterNode *node;

size_t pat_len;

char *pat, *cmd;

clusterReply *resp;

int i, cmd_len;

if (zend_parse_parameters(ZEND_NUM_ARGS(), "s", &pat, &pat_len)

== FAILURE)

{

RETURN_FALSE;

}

/* Prefix and then build our command */

cmd_len = redis_spprintf(c->flags, NULL, &cmd, "KEYS", "k", pat, pat_len);

array_init(return_value);

/* Treat as readonly */

c->readonly = CLUSTER_IS_ATOMIC(c);

/* Iterate over our known nodes */

ZEND_HASH_FOREACH_PTR(c->nodes, node) {

if (node == NULL) continue;

if (cluster_send_slot(c, node->slot, cmd, cmd_len, TYPE_MULTIBULK

) < 0)

{

php_error_docref(0, E_ERROR, "Can't send KEYS to %s:%d",

ZSTR_VAL(node->sock->host), node->sock->port);

zval_dtor(return_value);

efree(cmd);

RETURN_FALSE;

}

/* Ensure we can get a response */

resp = cluster_read_resp(c, 0);

if (!resp) {

php_error_docref(0, E_WARNING,

"Can't read response from %s:%d", ZSTR_VAL(node->sock->host),

node->sock->port);

continue;

}

/* Iterate keys, adding to our big array */

for(i = 0; i < resp->elements; i++) {

/* Skip non bulk responses, they should all be bulk */

if (resp->element[i]->type != TYPE_BULK) {

continue;

}

add_next_index_stringl(return_value, resp->element[i]->str,

resp->element[i]->len);

}

/* Free response, don't free data */

cluster_free_reply(resp, 1);

} ZEND_HASH_FOREACH_END();

efree(cmd);

}

/* }}} */

/* {{{ proto int RedisCluster::type(string key) */

PHP_METHOD(RedisCluster, type) {

CLUSTER_PROCESS_KW_CMD("TYPE", redis_key_cmd, cluster_type_resp, 1);

}

/* }}} */

/* {{{ proto string RedisCluster::pop(string key, [int count = 0]) */

PHP_METHOD(RedisCluster, lpop) {

CLUSTER_PROCESS_KW_CMD("LPOP", redis_pop_cmd, cluster_pop_resp, 0);

}

/* }}} */

/* {{{ proto string RedisCluster::rpop(string key, [int count = 0]) */

PHP_METHOD(RedisCluster, rpop) {

CLUSTER_PROCESS_KW_CMD("RPOP", redis_pop_cmd, cluster_pop_resp, 0);

}

/* }}} */

/* {{{ proto bool RedisCluster::lset(string key, long index, string val) */

PHP_METHOD(RedisCluster, lset) {

CLUSTER_PROCESS_KW_CMD("LSET", redis_key_long_val_cmd, cluster_bool_resp, 0);

}

/* }}} */

/* {{{ proto string RedisCluster::spop(string key) */

PHP_METHOD(RedisCluster, spop) {

if (ZEND_NUM_ARGS() == 1) {

CLUSTER_PROCESS_KW_CMD("SPOP", redis_key_cmd, cluster_bulk_resp, 0);

} else if (ZEND_NUM_ARGS() == 2) {

CLUSTER_PROCESS_KW_CMD("SPOP", redis_key_long_cmd, cluster_mbulk_resp, 0);

} else {

ZEND_WRONG_PARAM_COUNT();

}

}

/* }}} */

/* {{{ proto string|array RedisCluster::srandmember(string key, [long count]) */

PHP_METHOD(RedisCluster, srandmember) {

redisCluster *c = GET_CONTEXT();

cluster_cb cb;

char *cmd; int cmd_len; short slot;

short have_count;

/* Treat as readonly */

c->readonly = CLUSTER_IS_ATOMIC(c);

if (redis_srandmember_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, c->flags,

&cmd, &cmd_len, &slot, NULL, &have_count)

== FAILURE)

{

RETURN_FALSE;

}

if (cluster_send_command(c,slot,cmd,cmd_len) < 0 || c->err != NULL) {

efree(cmd);

RETURN_FALSE;

}

// Clean up command

efree(cmd);

cb = have_count ? cluster_mbulk_resp : cluster_bulk_resp;

if (CLUSTER_IS_ATOMIC(c)) {

cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, c, NULL);

} else {

void *ctx = NULL;

CLUSTER_ENQUEUE_RESPONSE(c, slot, cb, ctx);

RETURN_ZVAL(getThis(), 1, 0);

}

}

/* {{{ proto string RedisCluster::strlen(string key) */

PHP_METHOD(RedisCluster, strlen) {

CLUSTER_PROCESS_KW_CMD("STRLEN", redis_key_cmd, cluster_long_resp, 1);

}

/* {{{ proto long RedisCluster::lpush(string key, string val1, ... valN) */

PHP_METHOD(RedisCluster, lpush) {

CLUSTER_PROCESS_KW_CMD("LPUSH", redis_key_varval_cmd, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::rpush(string key, string val1, ... valN) */

PHP_METHOD(RedisCluster, rpush) {

CLUSTER_PROCESS_KW_CMD("RPUSH", redis_key_varval_cmd, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto array RedisCluster::blpop(string key1, ... keyN, long timeout) */

PHP_METHOD(RedisCluster, blpop) {

CLUSTER_PROCESS_KW_CMD("BLPOP", redis_blocking_pop_cmd, cluster_mbulk_resp, 0);

}

/* }}} */

/* {{{ proto array RedisCluster::brpop(string key1, ... keyN, long timeout */

PHP_METHOD(RedisCluster, brpop) {

CLUSTER_PROCESS_KW_CMD("BRPOP", redis_blocking_pop_cmd, cluster_mbulk_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::rpushx(string key, mixed value) */

PHP_METHOD(RedisCluster, rpushx) {

CLUSTER_PROCESS_KW_CMD("RPUSHX", redis_kv_cmd, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::lpushx(string key, mixed value) */

PHP_METHOD(RedisCluster, lpushx) {

CLUSTER_PROCESS_KW_CMD("LPUSHX", redis_kv_cmd, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::linsert(string k,string pos,mix pvt,mix val) */

PHP_METHOD(RedisCluster, linsert) {

CLUSTER_PROCESS_CMD(linsert, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto string RedisCluster::lindex(string key, long index) */

PHP_METHOD(RedisCluster, lindex) {

CLUSTER_PROCESS_KW_CMD("LINDEX", redis_key_long_cmd, cluster_bulk_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::lrem(string key, long count, string val) */

PHP_METHOD(RedisCluster, lrem) {

CLUSTER_PROCESS_CMD(lrem, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto string RedisCluster::rpoplpush(string key, string key) */

PHP_METHOD(RedisCluster, rpoplpush) {

CLUSTER_PROCESS_KW_CMD("RPOPLPUSH", redis_key_key_cmd, cluster_bulk_resp, 0);

}

/* }}} */

/* {{{ proto string RedisCluster::brpoplpush(string key, string key, long tm) */

PHP_METHOD(RedisCluster, brpoplpush) {

CLUSTER_PROCESS_CMD(brpoplpush, cluster_bulk_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::llen(string key) */

PHP_METHOD(RedisCluster, llen) {

CLUSTER_PROCESS_KW_CMD("LLEN", redis_key_cmd, cluster_long_resp, 1);

}

/* }}} */

/* {{{ proto long RedisCluster::scard(string key) */

PHP_METHOD(RedisCluster, scard) {

CLUSTER_PROCESS_KW_CMD("SCARD", redis_key_cmd, cluster_long_resp, 1);

}

/* }}} */

/* {{{ proto array RedisCluster::smembers(string key) */

PHP_METHOD(RedisCluster, smembers) {

CLUSTER_PROCESS_KW_CMD("SMEMBERS", redis_key_cmd, cluster_mbulk_resp, 1);

}

/* }}} */

/* {{{ proto long RedisCluster::sismember(string key) */

PHP_METHOD(RedisCluster, sismember) {

CLUSTER_PROCESS_KW_CMD("SISMEMBER", redis_kv_cmd, cluster_1_resp, 1);

}

/* }}} */

/* {{{ proto long RedisCluster::sadd(string key, string val1 [, ...]) */

PHP_METHOD(RedisCluster, sadd) {

CLUSTER_PROCESS_KW_CMD("SADD", redis_key_varval_cmd, cluster_long_resp, 0);

}

/* }}} */

/* {{{ proto long RedisCluster::saddarray(string key, array values) */

PHP_METHOD(RedisCluster, saddarray) {

CLUSTER_PROCESS_KW_CMD("SADD", redis_key_val_arr_cmd, cluster_long_resp, 0);

}

/* }}} */