Initial commit of save like commands

For commands like SAVE, BGSAVE, etc we need to change the
semantics from how you would call them directly, given that
we're communicating with multiple redis cluster nodes.

Each of these commands can be called in one of two ways.

1)  Pass a single argument, which will be treated as a key,
    and the command will be executed against the node where
    we believe that key would live.

2)  Pass a host and port, and as long as we know about this node
    the command will be forwarded there.

4 files changed, 162 insertions, 18 deletions

diff --git a/cluster_library.c b/cluster_library.c
index b71d85ef..577739b8 100644
--- a/cluster_library.c
+++ b/cluster_library.c
@@ -859,24 +859,6 @@ cluster_map_keyspace(redisCluster *cluster TSRMLS_DC) {
     return 0;
 }
 
-/* Helper to find if we've got a host:port mapped in our cluster nodes. */
-static redisClusterNode *cluster_find_node(redisCluster *c, const char *host,
-                                           unsigned short port)
-{
-    redisClusterNode **ret = NULL;
-    int key_len;
-    char key[1024];
-
-    key_len = snprintf(key,sizeof(key),"%s:%d", host, port);
-
-    if(zend_hash_find(c->nodes, key, key_len+1, (void**)&ret)==SUCCESS) {
-        return *ret;
-    }
-
-    // Not found
-    return NULL;
-}
-
 /* Parse the MOVED OR ASK redirection payload when we get such a response
  * and apply this information to our cluster.  If we encounter a parse error
  * nothing in the cluster will be modified, and -1 is returned. */
@@ -1057,6 +1039,24 @@ static int cluster_sock_write(redisCluster *c, unsigned short slot,
     return -1;
 }
 
+/* Helper to find if we've got a host:port mapped in our cluster nodes. */
+static redisClusterNode *cluster_find_node(redisCluster *c, const char *host,
+                                           unsigned short port)
+{
+    redisClusterNode **ret = NULL;
+    int key_len;
+    char key[1024];
+
+    key_len = snprintf(key,sizeof(key),"%s:%d", host, port);
+
+    if(zend_hash_find(c->nodes, key, key_len+1, (void**)&ret)==SUCCESS) {
+        return *ret;
+    }
+
+    // Not found
+    return NULL;
+}
+
 /* Provided a redisCluster object, the slot where we thought data was and
  * the slot where data was moved, update our node mapping */
 static void cluster_update_slot(redisCluster *c TSRMLS_CC) {
@@ -1186,6 +1186,27 @@ static int cluster_send_multi(redisCluster *c, short slot TSRMLS_DC) {
     return 0;
 }
 
+/* Iterate through our slots, looking for the host/port in question.  This 
+ * should perform well enough as in almost all situations, a few or a few
+ * dozen servers will map all the slots */
+PHPAPI short cluster_find_slot(redisCluster *c, const char *host,
+                               unsigned short port)
+{
+    int i;
+
+    for(i=0;i<REDIS_CLUSTER_SLOTS;i++) {
+        if(c->master[i] && c->master[i]->sock && 
+           c->master[i]->sock->port == port &&
+           !strcasecmp(c->master[i]->sock->host, host))
+        {
+            return i;
+        }
+    }
+
+    // We didn't find it
+    return -1;
+}
+
 /* Send a command to a specific slot */
 PHPAPI int cluster_send_slot(redisCluster *c, short slot, char *cmd, 
                              int cmd_len, REDIS_REPLY_TYPE rtype TSRMLS_DC)
diff --git a/cluster_library.h b/cluster_library.h
index f020d487..c02890c5 100644
--- a/cluster_library.h
+++ b/cluster_library.h
@@ -345,6 +345,9 @@ PHPAPI int cluster_send_discard(redisCluster *c, short slot TSRMLS_DC);
 PHPAPI int cluster_abort_exec(redisCluster *c TSRMLS_DC);
 PHPAPI int cluster_reset_multi(redisCluster *c);
 
+
+PHPAPI short cluster_find_slot(redisCluster *c, const char *host,
+    unsigned short port);
 PHPAPI int cluster_send_slot(redisCluster *c, short slot, char *cmd, 
     int cmd_len, REDIS_REPLY_TYPE rtype TSRMLS_DC);
 
diff --git a/redis_cluster.c b/redis_cluster.c
index c62e4dc6..ac0af13b 100644
--- a/redis_cluster.c
+++ b/redis_cluster.c
@@ -160,6 +160,14 @@ zend_function_entry redis_cluster_functions[] = {
     PHP_ME(RedisCluster, discard, NULL, ZEND_ACC_PUBLIC)
     PHP_ME(RedisCluster, watch, NULL, ZEND_ACC_PUBLIC)
     PHP_ME(RedisCluster, unwatch, NULL, ZEND_ACC_PUBLIC)
+
+    PHP_ME(RedisCluster, save, NULL, ZEND_ACC_PUBLIC)
+    PHP_ME(RedisCluster, bgsave, NULL, ZEND_ACC_PUBLIC)
+    PHP_ME(RedisCluster, flushdb, NULL, ZEND_ACC_PUBLIC)
+    PHP_ME(RedisCluster, flushall, NULL, ZEND_ACC_PUBLIC)
+    PHP_ME(RedisCluster, dbsize, NULL, ZEND_ACC_PUBLIC)
+    PHP_ME(RedisCluster, bgrewriteaof, NULL, ZEND_ACC_PUBLIC)
+    PHP_ME(RedisCluster, lastsave, NULL, ZEND_ACC_PUBLIC)
     {NULL, NULL, NULL}
 };
 
@@ -1732,4 +1740,107 @@ PHP_METHOD(RedisCluster, discard) {
     RETURN_TRUE;
 }
 
+/* Generic handler for things we want directed at a given node, like SAVE,
+ * BGSAVE, FLUSHDB, FLUSHALL, etc */
+static void 
+cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAMETERS, char *kw, 
+                       REDIS_REPLY_TYPE reply_type, cluster_cb cb)
+{
+    redisCluster *c = GET_CONTEXT();
+    char *cmd, *arg1; 
+    int arg1_len, cmd_len, arg1_free; 
+    short slot;
+    long arg2;
+
+    if(zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|l", &arg1, &arg1_len,
+                             &arg2)==FAILURE)
+    {
+        RETURN_FALSE;
+    }
+
+    // One argument means find the node (treated like a key), and two means
+    // send the command to a specific host and port
+    if(ZEND_NUM_ARGS() == 1) {
+        // Treat our argument like a key, and look for the slot that way
+        arg1_free = redis_key_prefix(c->flags, &arg1, &arg1_len);
+        slot = cluster_hash_key(arg1, arg1_len);
+        if(arg1_free) efree(arg1);        
+    } else {
+        // Find the slot by IP/port
+        slot = cluster_find_slot(c, (const char *)arg1, (unsigned short)arg2);
+        if(slot<0) {
+            php_error_docref(0 TSRMLS_CC, E_WARNING, "Unknown node %s:%ld", 
+                arg1, arg2);
+            RETURN_FALSE;
+        }
+    }
+
+    // Construct our command
+    cmd_len = redis_cmd_format_static(&cmd, kw, "");
+
+    // Kick off our command
+    if(cluster_send_slot(c, slot, cmd, cmd_len, reply_type TSRMLS_CC)<0) {
+        zend_throw_exception(redis_cluster_exception_ce,
+            "Unable to send command at a specific node", 0 TSRMLS_CC);
+        efree(cmd);
+        RETURN_FALSE;
+    }
+
+    // Our response callback
+    cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, c, NULL);
+
+    // Free our command
+    efree(cmd);
+}
+
+/* {{{ proto RedisCluster::save(string key)
+ *     proto RedisCluster::save(string host, long port) */
+PHP_METHOD(RedisCluster, save) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "SAVE", TYPE_LINE,
+        cluster_bool_resp);
+}
+
+/* {{{ proto RedisCluster::bgsave(string key) 
+ *     proto RedisCluster::bgsave(string host, long port) */
+PHP_METHOD(RedisCluster, bgsave) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "BGSAVE", 
+        TYPE_LINE, cluster_bool_resp);
+}
+
+/* {{{ proto RedisCluster::flushdb(string key)
+ *     proto RedisCluster::flushdb(string host, long port) */
+PHP_METHOD(RedisCluster, flushdb) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "FLUSHDB",
+        TYPE_LINE, cluster_bool_resp);
+}
+
+/* {{{ proto RedisCluster::flushall(string key)
+ *     proto RedisCluster::flushall(string host, long port) */
+PHP_METHOD(RedisCluster, flushall) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "FLUSHALL",
+        TYPE_LINE, cluster_bool_resp);
+}
+
+/* {{{ proto RedisCluster::dbsize(string key)
+ *     proto RedisCluster::dbsize(string host, long port) */
+PHP_METHOD(RedisCluster, dbsize) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "DBSIZE",
+        TYPE_LINE, cluster_bool_resp);
+}
+
+/* {{{ proto RedisCluster::bgrewriteaof(string key)
+ *     proto RedisCluster::bgrewriteaof(string host, long port) */
+PHP_METHOD(RedisCluster, bgrewriteaof) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "BGREWRITEAOF",
+        TYPE_LINE, cluster_bool_resp);
+}
+
+/* {{{ proto RedisCluster::lastsave(string key)
+ *     proto RedisCluster::lastsave(string host, long port) */
+PHP_METHOD(RedisCluster, lastsave) {
+    cluster_empty_node_cmd(INTERNAL_FUNCTION_PARAM_PASSTHRU, "LASTSAVE",
+        TYPE_INT, cluster_long_resp);
+}
+
+
 /* vim: set tabstop=4 softtabstops=4 noexpandtab shiftwidth=4: */
diff --git a/redis_cluster.h b/redis_cluster.h
index 45bc288b..321618a6 100644
--- a/redis_cluster.h
+++ b/redis_cluster.h
@@ -228,6 +228,15 @@ PHP_METHOD(RedisCluster, discard);
 PHP_METHOD(RedisCluster, watch);
 PHP_METHOD(RedisCluster, unwatch);
 
+/* DB saving, etc */
+PHP_METHOD(RedisCluster, save);
+PHP_METHOD(RedisCluster, bgsave);
+PHP_METHOD(RedisCluster, flushdb);
+PHP_METHOD(RedisCluster, flushall);
+PHP_METHOD(RedisCluster, dbsize);
+PHP_METHOD(RedisCluster, bgrewriteaof);
+PHP_METHOD(RedisCluster, lastsave);
+
 /* Introspection */
 PHP_METHOD(RedisCluster, getoption);
 PHP_METHOD(RedisCluster, setoption);