some minor bugs in ht_serialize. removed serialize test file

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) {

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");

modules_k/htable/ht_serialize_test.c

@@ -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));
-}

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;
 }