1 Jun
2011
1 Jun
'11
6:16 p.m.
Module: sip-router
Branch: mariusbucur/dmq
Commit: 3b45fb3e161872361a1efb1ea03f5e5e6a550c9c
URL:
http://git.sip-router.org/cgi-bin/gitweb.cgi/sip-router/?a=commit;h=3b45fb3…
Author: Marius Bucur <marius(a)marius-bucur.ro>
Committer: Marius Bucur <marius(a)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;
}