git:mariusbucur/dmq: some minor bugs in ht_serialize. removed serialize test file - sr-dev

1 Jun 2011

Module: sip-router
Branch: mariusbucur/dmq
Commit: 3b45fb3e161872361a1efb1ea03f5e5e6a550c9c
URL:    http://git.sip-router.org/cgi-bin/gitweb.cgi/sip-router/?a=commit;h=3b45fb3e...
Author: Marius Bucur marius@marius-bucur.ro
Committer: Marius Bucur marius@marius-bucur.ro
Date:   Wed Jun  1 19:16:22 2011 +0300
some minor bugs in ht_serialize. removed serialize test file
---
modules_k/dmq/worker.c               |    1 +
 modules_k/htable/ht_serialize.c      |    9 ++-
 modules_k/htable/ht_serialize_test.c |  108 ----------------------------------
 modules_k/htable/htable.c            |   62 +++++++++++++++++++
 4 files changed, 70 insertions(+), 110 deletions(-)

diff --git a/modules_k/dmq/worker.c b/modules_k/dmq/worker.c
index f88ac29..cf206f0 100644
--- a/modules_k/dmq/worker.c
+++ b/modules_k/dmq/worker.c
@@ -60,6 +60,7 @@ void worker_loop(int id) {
    			ret_value = current_job->f(current_job->msg, &peer_response);
    			if(ret_value < 0) {
    				LM_ERR("running job failed\n");
+					continue;
    			}
    			/* add the body to the reply */
    			if(peer_response.body.s) {
diff --git a/modules_k/htable/ht_serialize.c b/modules_k/htable/ht_serialize.c
index 6e046f7..713a126 100644
--- a/modules_k/htable/ht_serialize.c
+++ b/modules_k/htable/ht_serialize.c
@@ -7,6 +7,7 @@
 int serialize_ht_pair(pv_value_t* val, str* htname, str* s) {
    str encoded_val = {0, 0};
    str encoded_htname = {0, 0};
+	int len;
    if (!s) {
    	LM_ERR("no destination string given\n");
    	goto error;
@@ -18,17 +19,21 @@ int serialize_ht_pair(pv_value_t* val, str* htname, str* s) {
    if(val->rs.len) {
    	encoded_val.len = base64_enc_len(val->rs.len);
    	encoded_val.s = pkg_malloc(encoded_val.len);
-		if(base64_enc((unsigned char*)val->rs.s, val->rs.len, (unsigned char*)encoded_val.s, encoded_val.len) < 0) {
+		len = base16_enc((unsigned char*)val->rs.s, val->rs.len, (unsigned char*)encoded_val.s, encoded_val.len);
+		if(len < 0) {
    		LM_ERR("cannot encode value\n");
    		goto error;
    	}
+		encoded_val.len = len;
    }
    encoded_htname.len = base64_enc_len(htname->len);
    encoded_htname.s = pkg_malloc(encoded_htname.len);
-	if(base64_enc((unsigned char*)htname->s, htname->len, (unsigned char*)encoded_htname.s, encoded_htname.len) < 0) {
+	len = base16_enc((unsigned char*)htname->s, htname->len, (unsigned char*)encoded_htname.s, encoded_htname.len);
+	if(len < 0) {
    	LM_ERR("cannot encode htname\n");
    	goto error;
    }
+	encoded_htname.len = len;
    s->len = snprintf(s->s, s->len, "%d %d %.*s %.*s", val->flags, val->ri, STR_FMT(&encoded_htname), STR_FMT(&encoded_val));
    if(s->len < 0) {
    	LM_ERR("cannot serialize data - probably an small buffer\n");
diff --git a/modules_k/htable/ht_serialize_test.c b/modules_k/htable/ht_serialize_test.c
deleted file mode 100644
index 495cdcf..0000000
--- a/modules_k/htable/ht_serialize_test.c
+++ /dev/null
@@ -1,108 +0,0 @@
-#include <stdlib.h>
-#include <stdio.h>
-#include <string.h>
-#define LM_ERR printf
-#define pkg_malloc malloc
-#define pkg_free free
-#include "../../str.h"
-#include "../../basex.h"
-
-typedef struct _pv_value
-{
-	str rs;    /*!< string value */
-	int ri;    /*!< integer value */
-	int flags; /*!< flags about the type of value */
-} pv_value_t, *pv_value_p;
-
-/* snprintf - pretty ugly, but cds/serialize is unusable for the moment */
-int serialize_ht_pair(pv_value_t* val, str* htname, str* s) {
-	str encoded_val = {0, 0};
-	str encoded_htname = {0, 0};
-	if (!s) {
-		LM_ERR("no destination string given\n");
-		goto error;
-	}
-	if(!htname || !htname->s || !htname->len) {
-		LM_ERR("no hashtable name given\n");
-		goto error;
-	}
-	if(val->rs.len) {
-		encoded_val.len = base64_enc_len(val->rs.len);
-		encoded_val.s = pkg_malloc(encoded_val.len);
-		if(base64_enc((unsigned char*)val->rs.s, val->rs.len, (unsigned char*)encoded_val.s, encoded_val.len) < 0) {
-			LM_ERR("cannot encode value\n");
-			goto error;
-		}
-	}
-	encoded_htname.len = base64_enc_len(htname->len);
-	encoded_htname.s = pkg_malloc(encoded_htname.len);
-	if(base64_enc((unsigned char*)htname->s, htname->len, (unsigned char*)encoded_htname.s, encoded_htname.len) < 0) {
-		LM_ERR("cannot encode htname\n");
-		goto error;
-	}
-	s->len = snprintf(s->s, s->len, "%d %d %.*s %.*s", val->flags, val->ri, STR_FMT(&encoded_htname), STR_FMT(&encoded_val));
-	if(s->len < 0) {
-		LM_ERR("cannot serialize data - probably an small buffer\n");
-		goto error;
-	}
-	
-	if(encoded_val.s) {
-		pkg_free(encoded_val.s);
-	}
-	if(encoded_htname.s) {
-		pkg_free(encoded_htname.s);
-	}
-	return 0;
-error:
-	if(encoded_val.s) {
-		pkg_free(encoded_val.s);
-	}
-	if(encoded_htname.s) {
-		pkg_free(encoded_htname.s);
-	}
-	return -1;
-}
-
-int deserialize_ht_pair(pv_value_t* val, str* htname, str* src) {
-	str encoded_htname = {0, 0};
-	str encoded_val = {0, 0};
-	encoded_htname.s = pkg_malloc(src->len);
-	memset(encoded_htname.s, 0, src->len);
-	encoded_val.s = pkg_malloc(src->len);
-	memset(encoded_val.s, 0, src->len);
-	
-	sscanf(src->s, "%d %d %s %s", &val->flags, &val->ri, encoded_htname.s, encoded_val.s);
-	encoded_htname.len = strlen(encoded_htname.s);
-	encoded_val.len = strlen(encoded_val.s);
-	
-	if(base64_dec((unsigned char*)encoded_htname.s, encoded_htname.len, (unsigned char*)htname->s, htname->len) < 0) {
-		LM_ERR("cannot decode htname\n");
-		goto error;
-	}
-	if(base64_dec((unsigned char*)encoded_val.s, encoded_val.len, (unsigned char*)val->rs.s, val->rs.len) < 0) {
-		LM_ERR("cannot decode val\n");
-		goto error;
-	}
-	
-	pkg_free(encoded_htname.s);
-	pkg_free(encoded_val.s);
-	return 0;
-error:
-	pkg_free(encoded_htname.s);
-	pkg_free(encoded_val.s);
-	return -1;
-}
-
-int main(){
-	pv_value_t pv;
-	str hname = str_init("hashtable name");
-	str src;
-	src.len = 2048;
-	src.s = pkg_malloc(src.len);
-	pv.flags = 2;
-	pv.ri = 0;
-	pv.rs.len = 12;
-	pv.rs.s = "hello world";
-	printf("%d\n", serialize_ht_pair(&pv, &hname, &src));
-	printf("%.*s\n", STR_FMT(&src));
-}
\ No newline at end of file
diff --git a/modules_k/htable/htable.c b/modules_k/htable/htable.c
index f7af1d9..3b645a6 100644
--- a/modules_k/htable/htable.c
+++ b/modules_k/htable/htable.c
@@ -39,6 +39,10 @@
 #include "../../lib/kcore/faked_msg.h"
#include "../../pvar.h"
+#include "../dmq/dmq.h"
+#include "../../parser/msg_parser.h"
+#include "../../parser/parse_content.h"
+
 #include "ht_api.h"
 #include "ht_db.h"
 #include "ht_var.h"
@@ -47,7 +51,53 @@
MODULE_VERSION
+/* dmq API structure */
+dmq_api_t ht_dmq_bind;
+register_dmq_peer_t ht_register_dmq;
+dmq_peer_t* ht_dmq_peer;
+
+int dmq_htable_callback(struct sip_msg* msg, peer_reponse_t* resp) {
+	int content_length;
+	str body;
+	if(parse_headers(msg, HDR_EOH_F, 0) < 0) {
+		LM_ERR("error parsing message headers\n");
+		goto error;
+	}
+	if(!msg->content_length) {
+		LM_ERR("no content length header found\n");
+		goto error;
+	}
+	content_length = get_content_length(msg);
+	if(!content_length) {
+		LM_ERR("content length is 0\n");
+		goto error;
+	}
+	body.s = get_body(msg);
+	body.len = content_length;
+	LM_ERR("it worked - dmq module triggered the htable callback [%ld %d]\n", time(0), my_pid());
+	str ct = str_init("text/xml");
+	str reason = str_init("200 OK");
+	resp->content_type = ct;
+	resp->reason = reason;
+	resp->body.s = 0;
+	resp->resp_code = 200;
+	return 0;
+error:
+	return -1;
+}
+
+static void add_dmq_peer() {
+	dmq_peer_t htable_peer;
+	htable_peer.peer_id.s = "htable";
+	htable_peer.peer_id.len = 6;
+	htable_peer.description.s = "ditributed htable implementation using dmq";
+	htable_peer.description.len = 42;
+	htable_peer.callback = dmq_htable_callback;
+	ht_dmq_peer = ht_register_dmq(&htable_peer);
+}
+
 int  ht_timer_interval = 20;
+int ht_use_dmq = 0;
static int htable_init_rpc(void);
@@ -108,6 +158,7 @@ static param_export_t params[]={
    {"array_size_suffix",  STR_PARAM, &ht_array_size_suffix.s},
    {"fetch_rows",         INT_PARAM, &ht_fetch_rows},
    {"timer_interval",     INT_PARAM, &ht_timer_interval},
+	{"use_dmq",     INT_PARAM, &ht_use_dmq},
    {0,0,0}
 };
@@ -172,6 +223,17 @@ static int mod_init(void)
    		return -1;
    	}
    }
+	
+	if(ht_use_dmq){
+		if(dmq_load_api(&ht_dmq_bind) < 0) {
+			LM_ERR("cannot load dmq api\n");
+			return -1;
+		} else {
+			ht_register_dmq = ht_dmq_bind.register_dmq_peer;
+			add_dmq_peer();
+			LM_DBG("presence-dmq loaded\n");
+		}
+	}
    return 0;
 }

    