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