11 Sep
2014
11 Sep
'14
4:48 p.m.
Module: sip-router
Branch: lazedo/kazoo
Commit: 0d9cf1bd0b34f0fe718d6f52edf72b5a890b62fe
URL:
http://git.sip-router.org/cgi-bin/gitweb.cgi/sip-router/?a=commit;h=0d9cf1b…
Author: lazedo <luis.azedo(a)factorlusitano.com>
Committer: lazedo <luis.azedo(a)factorlusitano.com>
Date: Thu Sep 11 14:47:33 2014 +0100
support vhost in connection
---
modules/kazoo/kz_amqp.c | 59 ++++++++++++++++++++++++++++++++++++++++------
modules/kazoo/kz_amqp.h | 3 +-
2 files changed, 52 insertions(+), 10 deletions(-)
diff --git a/modules/kazoo/kz_amqp.c b/modules/kazoo/kz_amqp.c
index 1e3c0d2..3931d27 100644
--- a/modules/kazoo/kz_amqp.c
+++ b/modules/kazoo/kz_amqp.c
@@ -127,6 +127,17 @@ void kz_amqp_free_bind(kz_amqp_bind_ptr bind)
shm_free(bind);
}
+void kz_amqp_free_connection(kz_amqp_conn_ptr conn)
+{
+ if(!conn)
+ return;
+
+ if(conn->url)
+ shm_free(conn->url);
+ shm_free(conn);
+}
+
+
void kz_amqp_free_pipe_cmd(kz_amqp_cmd_ptr cmd)
{
if(cmd == NULL)
@@ -269,9 +280,9 @@ void kz_amqp_destroy() {
if(kz_pool != NULL) {
kz_amqp_conn_ptr conn = kz_pool->head;
while(conn != NULL) {
- kz_amqp_conn_ptr free = conn;
+ kz_amqp_conn_ptr tofree = conn;
conn = conn->next;
- shm_free(free);
+ kz_amqp_free_connection(tofree);
}
shm_free(kz_pool);
}
@@ -279,13 +290,45 @@ void kz_amqp_destroy() {
}
+#define KZ_URL_MAX_SIZE 50
+static char* KZ_URL_ROOT = "/";
+
int kz_amqp_add_connection(modparam_t type, void* val)
{
kz_amqp_init_connection_pool(); // find a better way
+ char* url = (char*) val;
+ int len = strlen(url);
+ if(len > KZ_URL_MAX_SIZE) {
+ LM_ERR("connection url exceeds max size %d\n", KZ_URL_MAX_SIZE);
+ return -1;
+ }
+
kz_amqp_conn_ptr newConn = shm_malloc(sizeof(kz_amqp_conn));
memset(newConn, 0, sizeof(kz_amqp_conn));
+ newConn->url = shm_malloc( (KZ_URL_MAX_SIZE + 1) * sizeof(char) );
+ memset(newConn->url, 0, (KZ_URL_MAX_SIZE + 1) * sizeof(char));
+ // maintain compatibility
+ if (!strncmp((char*)val, "kazoo://", 8)) {
+ sprintf(newConn->url, "amqp://%s", (char*)(url+(8*sizeof(char))) );
+ } else {
+ strcpy(newConn->url, url);
+ newConn->url[len] = '\0';
+ }
+
+ if(amqp_parse_url(newConn->url, &newConn->info) == AMQP_STATUS_BAD_URL) {
+ LM_ERR("ERROR PARSING URL \"%s\"\n", newConn->url);
+ goto error;
+ }
+
+
+ if(newConn->info.vhost == NULL) {
+ newConn->info.vhost = KZ_URL_ROOT;
+ } else if(newConn->info.vhost[0] == '/' &&
strlen(newConn->info.vhost) == 1) { // bug in amqp_parse_url ?
+ newConn->info.vhost++;
+ }
+
if(kz_pool->head == NULL)
kz_pool->head = newConn;
@@ -294,9 +337,12 @@ int kz_amqp_add_connection(modparam_t type, void* val)
kz_pool->tail = newConn;
- amqp_parse_url((char*)val, &newConn->info);
-
return 0;
+
+error:
+ kz_amqp_free_connection(newConn);
+ return -1;
+
}
void kz_amqp_connection_close(kz_amqp_conn_ptr rmq) {
@@ -313,9 +359,6 @@ void kz_amqp_connection_close(kz_amqp_conn_ptr rmq) {
rmq->conn = NULL;
rmq->socket = NULL;
rmq->channel_count = 0;
-
-// lock_release(&kz_pool->lock);
-
}
}
@@ -348,7 +391,7 @@ int kz_amqp_connection_open(kz_amqp_conn_ptr rmq) {
}
if (kz_amqp_error("Logging in", amqp_login(rmq->conn,
- "/", //rmq->info.vhost,
+ rmq->info.vhost,
0,
131072,
0,
diff --git a/modules/kazoo/kz_amqp.h b/modules/kazoo/kz_amqp.h
index ce4aa5d..10b9095 100644
--- a/modules/kazoo/kz_amqp.h
+++ b/modules/kazoo/kz_amqp.h
@@ -30,11 +30,11 @@ extern int dbk_consumer_processes;
typedef struct kz_amqp_conn_t {
kz_amqp_connection_info info;
+ char* url;
amqp_connection_state_t conn;
amqp_socket_t *socket;
amqp_channel_t channel_count;
amqp_channel_t channel_counter;
-// gen_lock_t lock;
struct kz_amqp_conn_t* next;
} kz_amqp_conn, *kz_amqp_conn_ptr;
@@ -42,7 +42,6 @@ typedef struct {
kz_amqp_conn_ptr current;
kz_amqp_conn_ptr head;
kz_amqp_conn_ptr tail;
-// gen_lock_t lock;
} kz_amqp_conn_pool, *kz_amqp_conn_pool_ptr;