improvements regarding the notification system

modules_k/dmq/dmq.c

@@ -42,6 +42,7 @@
 #include "../../mem/shm_mem.h"
 #include "../../usr_avp.h"
 #include "../../modules/tm/tm_load.h"
+#include "../../parser/parse_uri.h"
 #include "../../modules/sl/sl.h"
 #include "../../pt.h"
 #include "../../lib/kmi/mi.h"
@@ -66,6 +67,10 @@ int pid = 0;
 /* module parameters */
 int num_workers = DEFAULT_NUM_WORKERS;
 str dmq_server_address = {0, 0};
+struct sip_uri dmq_server_uri;
+
+str dmq_notification_address = {0, 0};
+struct sip_uri dmq_notification_uri;
 
 /* TM bind */
 struct tm_binds tmb;
@@ -86,7 +91,7 @@ static int mod_init(void);
 static int child_init(int);
 static void destroy(void);
 static int handle_dmq_fixup(void** param, int param_no);
-static int check_dmq_server_address();
+static int parse_server_address(str* uri, struct sip_uri* parsed_uri);
 
 static cmd_export_t cmds[] = {
 	{"handle_dmq_message",  (cmd_function)handle_dmq_message, 0, handle_dmq_fixup, 0, 
@@ -99,6 +104,7 @@ static cmd_export_t cmds[] = {
 static param_export_t params[] = {
 	{"num_workers", INT_PARAM, &num_workers},
 	{"server_address", STR_PARAM, &dmq_server_address.s},
+	{"notification_address", STR_PARAM, &dmq_notification_address.s},
 	{0, 0, 0}
 };
 
@@ -152,13 +158,18 @@ static int mod_init(void) {
 	
 	/* register worker processes - add one because of the ping process */
 	register_procs(num_workers);
-	/* check server_address not empty and correct */
 	
-	if(check_dmq_server_address() < 0) {
+	/* check server_address and notification_address are not empty and correct */
+	if(parse_server_address(&dmq_server_address, &dmq_server_uri) < 0) {
 		LM_ERR("server address invalid\n");
 		return -1;
 	}
 	
+	if(parse_server_address(&dmq_notification_address, &dmq_notification_uri) < 0) {
+		LM_ERR("notification address invalid\n");
+		return -1;
+	}
+	
 	/* allocate workers array */
 	workers = shm_malloc(num_workers * sizeof(*workers));
 	if(workers == NULL) {
@@ -213,27 +224,25 @@ static int child_init(int rank) {
  * destroy function
  */
 static void destroy(void) {
+	/* TODO unregister dmq node, free resources */
 }
 
 static int handle_dmq_fixup(void** param, int param_no) {
  	return 0;
 }
 
-static int check_dmq_server_address() {
-	if(!dmq_server_address.s) {
+static int parse_server_address(str* uri, struct sip_uri* parsed_uri) {
+	if(!uri->s) {
+		LM_ERR("server address missing\n");
 		return -1;
 	}
-	dmq_server_address.len = strlen(dmq_server_address.s);
-	if(!dmq_server_address.len) {
+	uri->len = strlen(uri->s);
+	if(!uri->len) {
 		LM_ERR("empty server address\n");
 		return -1;
 	}
-	if(strncmp(dmq_server_address.s, "sip:", 4)) {
-		LM_ERR("server address must start with sip:\n");
-		return -1;
-	}
-	if(!strchr(dmq_server_address.s + 4, ':')) {
-		LM_ERR("server address must be of form sip:host:port\n");
+	if(parse_uri(uri->s, uri->len, parsed_uri) < 0) {
+		LM_ERR("error parsing server address\n");
 		return -1;
 	}
 	return 0;

modules_k/dmq/dmq.h

@@ -5,6 +5,7 @@
 #include "../../error.h"
 #include "../../sr_module.h"
 #include "../../modules/tm/tm_load.h"
+#include "../../parser/parse_uri.h"
 #include "../../modules/sl/sl.h"
 #include "bind_dmq.h"
 #include "peer.h"
@@ -20,7 +21,9 @@ extern str dmq_server_address;
 extern dmq_peer_list_t* peer_list;
 extern dmq_node_list_t* node_list;
 extern str dmq_request_method;
-
+extern struct sip_uri dmq_server_uri;
+extern str dmq_notification_address;
+extern struct sip_uri dmq_notification_uri;
 /* sl and tm */
 extern struct tm_binds tmb;
 extern sl_api_t slb;

modules_k/dmq/dmq_funcs.c

@@ -16,18 +16,63 @@ void dmq_tm_callback( struct cell *t, int type, struct tmcb_params *ps) {
 	LM_ERR("callback\n");
 }
 
-int send_dmq_message(dmq_peer_t* peer, str* body) {
+int build_from_str(str* username, struct sip_uri* uri, str* from) {
+	/* sip:user@host:port */
+	int from_len = username->len + uri->host.len + uri->port.len + 10;
+	if(!uri->host.s || !uri->host.len) {
+		LM_ERR("no host in uri\n");
+		return -1;
+	}
+	if(!username->s || !username->len) {
+		LM_ERR("no username given\n");
+		return -1;
+	}
+	from->s = pkg_malloc(from_len);
+	from->len = 0;
+	
+	memcpy(from->s, "sip:", 4);
+	from->len += 4;
+	
+	memcpy(from->s + from->len, username->s, username->len);
+	from->len += username->len;
+	
+	memcpy(from->s + from->len, "@", 1);
+	from->len += 1;
+	
+	memcpy(from->s + from->len, uri->host.s, uri->host.len);
+	from->len += uri->host.len;
+	
+	if(uri->port.s && uri->port.len) {
+		memcpy(from->s + from->len, ":", 1);
+		from->len += 1;
+		memcpy(from->s + from->len, uri->port.s, uri->port.len);
+		from->len += uri->port.len;
+	}
+	return 0;
+}
+
+int send_dmq_message(dmq_peer_t* peer, str* body, dmq_node_t* node) {
 	uac_req_t uac_r;
 	str str_hdr = {0, 0};
 	/* TODO - do not hardcode these - just for tesing purposes */
-	str from = {"sip:[email protected]:5060", 25};
-	str req_uri = from;
+	str from, to, req_uri;
 	void *cb_param = NULL;
 	int result;
+	
+	if(build_from_str(&peer->peer_id, &dmq_server_uri, &from) < 0) {
+		LM_ERR("error building from string\n");
+		return -1;
+	}
+	if(build_from_str(&peer->peer_id, &node->uri, &to) < 0) {
+		LM_ERR("error building to string\n");
+		return -1;
+	}
+	req_uri = to;
+	
 	set_uac_req(&uac_r, &dmq_request_method, &str_hdr, body, NULL, TMCB_LOCAL_COMPLETED,
 			dmq_tm_callback, (void*)cb_param);
 	result = tmb.t_request(&uac_r, &req_uri,
-			       &from, &from,
+			       &to, &from,
 			       NULL);
 	if(result < 0)
 	{

modules_k/dmq/dmq_funcs.h

@@ -9,6 +9,6 @@
 #include "../../str.h"
 
 int register_dmq_peer(dmq_peer_t* peer);
-int send_dmq_message(dmq_peer_t* peer, str* body);
+int send_dmq_message(dmq_peer_t* peer, str* body, dmq_node_t* node);
 
 #endif

modules_k/dmq/dmqnode.c

@@ -2,6 +2,9 @@
 #include "dmqnode.h"
 #include "dmq.h"
 
+dmq_node_t* self_node;
+dmq_node_t* notification_node;	
+
 dmq_node_list_t* init_dmq_node_list() {
 	dmq_node_list_t* node_list = shm_malloc(sizeof(dmq_node_list_t));
 	memset(node_list, 0, sizeof(dmq_node_list_t));
@@ -9,14 +12,21 @@ dmq_node_list_t* init_dmq_node_list() {
 	return node_list;
 }
 
-inline int add_dmq_node(dmq_node_list_t* list, dmq_node_t* newnode) {
-	dmq_node_t* copy = shm_malloc(sizeof(dmq_node_t));
-	memcpy(copy, newnode, sizeof(dmq_node_t));
-	shm_str_dup(&copy->orig_uri, &newnode->orig_uri);
+inline dmq_node_t* add_dmq_node(dmq_node_list_t* list, str* uri) {
+	dmq_node_t* newnode = shm_malloc(sizeof(dmq_node_t));
+	memset(newnode, 0, sizeof(dmq_node_t));
+	shm_str_dup(&newnode->orig_uri, uri);
+	if(parse_uri(newnode->orig_uri.s, newnode->orig_uri.len, &newnode->uri) < 0) {
+		LM_ERR("error in parsing node uri\n");
+		goto error;
+	}
 	lock_get(&list->lock);
-	copy->next = list->nodes;
-	list->nodes = copy;
+	newnode->next = list->nodes;
+	list->nodes = newnode;
 	list->count++;
 	lock_release(&list->lock);
+	return newnode;
+error:
+	shm_free(newnode);
 	return 0;
 }

modules_k/dmq/dmqnode.h

@@ -13,7 +13,7 @@
 
 typedef struct dmq_node {
 	str orig_uri;
-	struct sip_uri* uri;
+	struct sip_uri uri;
 	int status;
 	int last_notification;
 	struct dmq_node* next;
@@ -27,6 +27,9 @@ typedef struct dmq_node_list {
 
 dmq_node_list_t* init_dmq_node_list();
 int update_node_list(dmq_node_list_t* remote_list);
-int add_dmq_node(dmq_node_list_t* list, dmq_node_t* newnode);
+dmq_node_t* add_dmq_node(dmq_node_list_t* list, str* uri);
+
+extern dmq_node_t* self_node;
+extern dmq_node_t* notification_node;	
 
 #endif

modules_k/dmq/message.c

@@ -3,10 +3,16 @@
 #include "../../parser/parse_content.h"
 #include "../../parser/parse_from.h"
 #include "../../ut.h"
+#include "dmq.h"
 #include "worker.h"
 #include "peer.h"
 #include "message.h"
 
+static str dmq_200_rpl  = str_init("OK");
+static str dmq_400_rpl  = str_init("Bad request");
+static str dmq_500_rpl  = str_init("Server Internal Error");
+static str dmq_404_rpl  = str_init("User Not Found");
+
 int handle_dmq_message(struct sip_msg* msg, char* str1, char* str2) {
 	dmq_peer_t* peer;
 	if ((parse_sip_msg_uri(msg) < 0) || (!msg->parsed_uri.user.s)) {
@@ -19,11 +25,18 @@ int handle_dmq_message(struct sip_msg* msg, char* str1, char* str2) {
 	       ZSW(str1), ZSW(str2));
 	/* the peer id is given as the userinfo part of the request URI */
 	peer = find_peer(msg->parsed_uri.user);
-	if(!peer) {
+	if(peer) {
 		LM_DBG("no peer found for %.*s\n", msg->parsed_uri.user.len, msg->parsed_uri.user.s);
+		if(slb.freply(msg, 404, &dmq_404_rpl) < 0)
+		{
+			LM_ERR("sending reply\n");
+			goto error;
+		}
 		return 0;
 	}
 	LM_DBG("handle_dmq_message peer found: %.*s\n", msg->parsed_uri.user.len, msg->parsed_uri.user.s);
 	add_dmq_job(msg, peer);
 	return 0;
+error:
+	return -1;
 }

modules_k/dmq/notification_peer.c

@@ -2,23 +2,28 @@
 #include "dmq_funcs.h"
 
 int add_notification_peer() {
-	dmq_node_t self_node;
-	self_node.orig_uri = dmq_server_address;
-	
 	dmq_notification_peer.callback = dmq_notification_callback;
-	dmq_notification_peer.description.s = "dmqpeer";
-	dmq_notification_peer.description.len = 7;
-	dmq_notification_peer.peer_id.s = "dmqpeer";
-	dmq_notification_peer.peer_id.len = 7;
+	dmq_notification_peer.description.s = "notification_peer";
+	dmq_notification_peer.description.len = 17;
+	dmq_notification_peer.peer_id.s = "notification_peer";
+	dmq_notification_peer.peer_id.len = 17;
 	if(register_dmq_peer(&dmq_notification_peer) < 0) {
 		LM_ERR("error in register_dmq_peer\n");
 		goto error;
 	}
 	/* add itself to the node list */
-	if(add_dmq_node(node_list, &self_node) < 0) {
+	self_node = add_dmq_node(node_list, &dmq_server_address);
+	if(!self_node) {
 		LM_ERR("error adding self node\n");
 		goto error;
 	}
+	/* add the notification server to the node list */
+	notification_node = add_dmq_node(node_list, &dmq_notification_address);
+	if(!notification_node) {
+		LM_ERR("error adding notification node\n");
+		goto error;
+	}
+	/* initial notification request to receive the complete node list */
 	if(request_initial_nodelist() < 0) {
 		LM_ERR("error in request_initial_notification\n");
 		goto error;
@@ -28,6 +33,19 @@ error:
 	return -1;
 }
 
+/**
+ * extract the node list from the body of a notification request SIP message
+ * the SIP request will look something like:
+ * 	KDMQ sip:10.0.0.0:5062
+ * 	To: ...
+ * 	From: ...
+ * 	Max-Forwards: ...
+ * 	Content-Length: 22
+ * 	
+ * 	sip:host1:port1;param1=value1
+ * 	sip:host2:port2;param2=value2
+ * 	...
+ */
 int extract_node_list(dmq_node_list_t* update_list, struct sip_msg* msg) {
 	int content_length, total_nodes = 0;
 	str body;
@@ -100,7 +118,8 @@ int build_node_str(dmq_node_t* node, char* buf, int buflen) {
 	return node->orig_uri.len;
 }
 
-/* builds the body of a notification message from the list of servers 
+/**
+ * builds the body of a notification message from the list of servers 
  * the result will look something like:
  * sip:host1:port1;param1=value1\r\n
  * sip:host2:port2;param2=value2\r\n
@@ -146,7 +165,7 @@ error:
 int request_initial_nodelist() {
 	str* body = build_notification_body();
 	int ret;
-	ret = send_dmq_message(&dmq_notification_peer, body);
+	ret = send_dmq_message(&dmq_notification_peer, body, notification_node);
 	pkg_free(body->s);
 	pkg_free(body);
 	return ret;