ndb_redis: support for redis array reply

- replies from redis holding arrays can be accessed now from config
  file, the values in array being accessible via index
- the size of the array is give as a new property of the reply
- patch by Vicente Hernando

modules/ndb_redis/README

@@ -119,6 +119,14 @@ modparam("ndb_redis", "server", "name=srvX;addr=127.0.0.2;port=6379;db=4")
    returned by REDIS server; info - in case of error from REDIS, it will
    contain an info message.
 
+   If reply type is an array (as in hiredis.h), there are other keys
+   available:
+     * size - returns number of elements in the array.
+     * type[n] - returns the type of the nth element in the array. type -
+       returns array type.
+     * value[n] - returns value of the nth element. value - returns null
+       for an array. You need to get each element by index.
+
    Example 1.2. redis_cmd usage
 ...
 if(redis_cmd("srvN", "INCR cnt", "r")) {
@@ -133,4 +141,10 @@ redis_cmd("srvN", "SET ruri $ru", "r");
 
 # get a value
 redis_cmd("srvN", "GET foo", "r");
+
+# array example
+if(redis_cmd("srvN", "HMGET foo_key field1 field3", "r")) {
+    xlog("array size: $redis(r=>size)\n");
+    xlog("first values: $redis(r=>value[0]) , $redis(r=>value[1])\n");
+}
 ...

modules/ndb_redis/doc/ndb_redis_admin.xml

@@ -109,6 +109,29 @@ modparam("ndb_redis", "server", "name=srvX;addr=127.0.0.2;port=6379;db=4")
 			returned by REDIS server; info - in case of error from REDIS, it will
 			contain an info message.
 		</para>
+		<para>
+			If reply type is an array (as in hiredis.h), there are other keys
+			available:
+			<itemizedlist>
+			<listitem>
+				<para>
+				size - returns number of elements in the array.
+				</para>
+			</listitem>
+			<listitem>
+				<para>
+				type[n] - returns the type of the nth element in the array. type
+				- returns array type.
+				</para>
+			</listitem>
+			<listitem>
+				<para>
+				value[n] - returns value of the nth element. value - returns null
+				for an array. You need to get each element by index.
+				</para>
+			</listitem>
+			</itemizedlist>
+		</para>
 		<example>
 		<title><function>redis_cmd</function> usage</title>
 		<programlisting format="linespecific">
@@ -125,6 +148,12 @@ redis_cmd("srvN", "SET ruri $ru", "r");
 
 # get a value
 redis_cmd("srvN", "GET foo", "r");
+
+# array example
+if(redis_cmd("srvN", "HMGET foo_key field1 field3", "r")) {
+    xlog("array size: $redis(r=>size)\n");
+    xlog("first values: $redis(r=>value[0]) , $redis(r=>value[1])\n");
+}
 ...
 </programlisting>
 	    </example>

modules/ndb_redis/ndb_redis_mod.c

@@ -3,6 +3,9 @@
  *
  * Copyright (C) 2011 Daniel-Constantin Mierla (asipto.com)
  *
+ * Copyright (C) 2012 Vicente Hernando Ara (System One: www.systemonenoc.com)
+ *     - for: redis array reply support
+ *
  * This file is part of Kamailio, a free SIP server.
  *
  * Kamailio is free software; you can redistribute it and/or modify
@@ -24,6 +27,7 @@
 #include <stdio.h>
 #include <unistd.h>
 #include <stdlib.h>
+#include <ctype.h>
 
 #include "../../sr_module.h"
 #include "../../mem/mem.h"
@@ -293,6 +297,79 @@ int redis_srv_param(modparam_t type, void *val)
 }
 
 
+/**
+ *
+ */
+int redis_parse_index(str *in, gparam_t *gp)
+{
+	if(in->s[0]==PV_MARKER)
+	{
+		gp->type = GPARAM_TYPE_PVS;
+		gp->v.pvs = (pv_spec_t*)pkg_malloc(sizeof(pv_spec_t));
+		if (gp->v.pvs == NULL)
+		{
+			LM_ERR("no pkg memory left for pv_spec_t\n");
+			pkg_free(gp);
+			return -1;
+		}
+
+		if(pv_parse_spec(in, gp->v.pvs)==NULL)
+		{
+			LM_ERR("invalid PV identifier\n");
+			pkg_free(gp->v.pvs);
+			pkg_free(gp);
+			return -1;
+		}
+	} else {
+		gp->type = GPARAM_TYPE_INT;
+		if(str2sint(in, &gp->v.i) != 0)
+		{
+			LM_ERR("bad number <%.*s>\n", in->len, in->s);
+			return -1;
+		}
+	}
+	return 0;
+}
+
+
+/**
+ *
+ */
+int redis_parse_token(str *in, gparam_t *gp, int i)
+{
+	str tok;
+
+	while(i<in->len && isspace(in->s[i]))
+		i++;
+
+	if(i>=in->len-2)
+		return -1;
+
+	if(in->s[i++]!='[')
+		return -1;
+
+	while(i<in->len-1 && isspace(in->s[i]))
+		i++;
+	if(i==in->len-1 || in->s[i]==']')
+		return -1;
+	tok.s = &(in->s[i]);
+
+	while(i<in->len && !isspace(in->s[i]) && in->s[i]!=']')
+		i++;
+	if(i==in->len)
+		return -1;
+	tok.len = &(in->s[i]) - tok.s;
+	if(redis_parse_index(&tok, gp)!=0)
+		return -1;
+
+	while(i<in->len && isspace(in->s[i]))
+		i++;
+	if(i==in->len || in->s[i]!=']')
+		return -1;
+
+	return 0;
+}
+
 /**
  *
  */
@@ -317,46 +394,76 @@ static int pv_parse_redisc_name(pv_spec_p sp, str *in)
 	rpv->rname.s = pvs.s;
 	for(i=0; i<pvs.len-2; i++)
 	{
-		if(pvs.s[i]=='=')
-		{
-			if(pvs.s[i+1]!='>')
-			{
-				LM_ERR("invalid var spec [%.*s]\n",
-						in->len, in->s);
-				pkg_free(rpv);
-				return -1;
-			}
+		if(isspace(pvs.s[i]) || pvs.s[i]=='=') {
 			rpv->rname.len = i;
 			break;
 		}
 	}
+	rpv->rname.len = i;
 
 	if(rpv->rname.len==0)
-	{
-		LM_ERR("invalid var spec [%.*s]\n", in->len, in->s);
-		pkg_free(rpv);
-		return -1;
-	}
+		goto error_var;
+
+	while(i<pvs.len-2 && isspace(pvs.s[i]))
+		i++;
+
+	if(pvs.s[i]!='=')
+		goto error_var;
+
+	if(pvs.s[i+1]!='>')
+		goto error_var;
+
 	i += 2;
+	while(i<pvs.len && isspace(pvs.s[i]))
+		i++;
+
+	if(i>=pvs.len)
+		goto error_key;
+
 	rpv->rkey.s   = pvs.s + i;
 	rpv->rkey.len = pvs.len - i;
 
-	if(rpv->rkey.len==5 && strncmp(rpv->rkey.s, "value", 5)==0) {
+	/* Default pos param initialization. */
+	rpv->pos.type = GPARAM_TYPE_INT;
+	rpv->pos.v.i = -1;
+
+	if(rpv->rkey.len>=5 && strncmp(rpv->rkey.s, "value", 5)==0) {
 		rpv->rkeyid = 1;
-	} else if(rpv->rkey.len==4 && strncmp(rpv->rkey.s, "type", 4)==0) {
+		if(rpv->rkey.len>5)
+		{
+			i+=5;
+			if(redis_parse_token(&pvs, &(rpv->pos), i)!=0)
+				goto error_key;
+		}
+	} else if(rpv->rkey.len>=4 && strncmp(rpv->rkey.s, "type", 4)==0) {
 		rpv->rkeyid = 0;
+		if(rpv->rkey.len>4)
+		{
+			i+=4;
+			if(redis_parse_token(&pvs, &(rpv->pos), i)!=0)
+				goto error_key;
+		}
 	} else if(rpv->rkey.len==4 && strncmp(rpv->rkey.s, "info", 4)==0) {
 		rpv->rkeyid = 2;
+	} else if(rpv->rkey.len==4 && strncmp(rpv->rkey.s, "size", 4)==0) {
+		rpv->rkeyid = 3;
 	} else {
-		LM_ERR("invalid key spec in [%.*s]\n", in->len, in->s);
-		pkg_free(rpv);
-		return -1;
+		goto error_key;
 	}
 
 	sp->pvp.pvn.u.dname = (void*)rpv;
 	sp->pvp.pvn.type = PV_NAME_OTHER;
 	return 0;
 
+error_var:
+	LM_ERR("invalid var spec [%.*s]\n", in->len, in->s);
+	pkg_free(rpv);
+	return -1;
+
+error_key:
+	LM_ERR("invalid key spec in [%.*s]\n", in->len, in->s);
+	pkg_free(rpv);
+	return -1;
 }
 
 /**
@@ -367,6 +474,7 @@ static int pv_get_redisc(struct sip_msg *msg,  pv_param_t *param,
 {
 	redisc_pv_t *rpv;
 	str s;
+	int pos;
 
 	rpv = (redisc_pv_t*)param->pvn.u.dname;
 	if(rpv->reply==NULL)
@@ -379,27 +487,72 @@ static int pv_get_redisc(struct sip_msg *msg,  pv_param_t *param,
 	if(rpv->reply->rplRedis==NULL)
 		return pv_get_null(msg, param, res);
 
+
+	if(fixup_get_ivalue(msg, &rpv->pos, &pos)!=0)
+		return pv_get_null(msg, param, res);
+
 	switch(rpv->rkeyid) {
 		case 1:
+			/* value */
 			switch(rpv->reply->rplRedis->type) {
 				case REDIS_REPLY_STRING:
+					if(pos!=-1)
+						return pv_get_null(msg, param, res);
 					s.len = rpv->reply->rplRedis->len;
 					s.s = rpv->reply->rplRedis->str;
 					return pv_get_strval(msg, param, res, &s);
 				case REDIS_REPLY_INTEGER:
+					if(pos!=-1)
+						return pv_get_null(msg, param, res);
 					return pv_get_sintval(msg, param, res,
-							(int)rpv->reply->rplRedis->integer);
+										  (int)rpv->reply->rplRedis->integer);
+				case REDIS_REPLY_ARRAY:
+					if(pos<0 || pos>=(int)rpv->reply->rplRedis->elements)
+						return pv_get_null(msg, param, res);
+					if(rpv->reply->rplRedis->element[pos]==NULL)
+						return pv_get_null(msg, param, res);
+					switch(rpv->reply->rplRedis->element[pos]->type) {
+						case REDIS_REPLY_STRING:
+						s.len = rpv->reply->rplRedis->element[pos]->len;
+							s.s = rpv->reply->rplRedis->element[pos]->str;
+							return pv_get_strval(msg, param, res, &s);
+						case REDIS_REPLY_INTEGER:
+							return pv_get_sintval(msg, param, res,
+												  (int)rpv->reply->rplRedis->element[pos]->integer);
+						default:
+							return pv_get_null(msg, param, res);
+					}
 				default:
 					return pv_get_null(msg, param, res);
 			}
 		case 2:
+			/* info */
 			if(rpv->reply->rplRedis->str==NULL)
 				return pv_get_null(msg, param, res);
 			s.len = rpv->reply->rplRedis->len;
 			s.s = rpv->reply->rplRedis->str;
 			return pv_get_strval(msg, param, res, &s);
+		case 3:
+			/* size */
+			if(rpv->reply->rplRedis->type == REDIS_REPLY_ARRAY) {
+				return pv_get_uintval(msg, param, res, (unsigned int)rpv->reply->rplRedis->elements);
+			} else {
+				return pv_get_null(msg, param, res);
+			}
+		case 0:
+			/* type */
+			if(pos==-1)
+				return pv_get_sintval(msg, param, res,
+									  rpv->reply->rplRedis->type);
+			if(rpv->reply->rplRedis->type != REDIS_REPLY_ARRAY)
+				return pv_get_null(msg, param, res);
+			if(pos<0 || pos>=(int)rpv->reply->rplRedis->elements)
+				return pv_get_null(msg, param, res);
+			if(rpv->reply->rplRedis->element[pos]==NULL)
+				return pv_get_null(msg, param, res);
+			return pv_get_sintval(msg, param, res, rpv->reply->rplRedis->element[pos]->type);
 		default:
-			return pv_get_sintval(msg, param, res,
-					rpv->reply->rplRedis->type);
+			/* We do nothing. */
+			return pv_get_null(msg, param, res);
 	}
 }

modules/ndb_redis/redis_client.h

@@ -28,6 +28,7 @@
 
 #include "../../str.h"
 #include "../../parser/parse_param.h"
+#include "../../mod_fix.h"
 
 int redisc_init(void);
 int redisc_destroy(void);
@@ -55,6 +56,7 @@ typedef struct redisc_pv {
 	redisc_reply_t *reply;
 	str rkey;
 	int rkeyid;
+	gparam_t pos;  /* Array element position. */
 } redisc_pv_t;
 
 redisc_reply_t *redisc_get_reply(str *name);