Merge pull request #1402 from kamailio/cchance/presence_dmq

presence: dmq integration

src/lib/srdb1/schema/pr_presentity.xml

@@ -9,7 +9,7 @@
 
 <table id="presentity" xmlns:db="http://docbook.org/ns/docbook">
     <name>presentity</name>
-    <version>4</version>
+    <version>5</version>
     <type db="mysql">&MYSQL_TABLE_TYPE;</type>
     <description>
 		<db:para>
@@ -98,6 +98,14 @@
         <description>Priority of the record</description>
 	</column>
 
+    <column id="ruid">
+        <name>ruid</name>
+        <type>string</type>
+        <size>64</size>
+        <null/>
+        <description>Record internal unique id</description>
+    </column>
+
     <index>
         <name>presentity_idx</name>
         <colref linkend="username"/>
@@ -107,6 +115,12 @@
         <unique/>
     </index>
 
+    <index>
+        <name>ruid_idx</name>
+        <colref linkend="ruid"/>
+        <unique/>
+    </index>
+
     <index>
         <name>presentity_expires</name>
         <colref linkend="expires"/>

src/modules/dmq/dmq.c

@@ -349,7 +349,7 @@ static void dmq_rpc_list_nodes(rpc_t *rpc, void *c)
 			goto error;
 		if(rpc->struct_add(h, "SSsSdd", "host", &cur->uri.host, "port",
 				   &cur->uri.port, "resolved_ip", ip, "status",
-				   get_status_str(cur->status), "last_notification",
+				   dmq_get_status_str(cur->status), "last_notification",
 				   cur->last_notification, "local", cur->local)
 				< 0)
 			goto error;

src/modules/dmq/dmqnode.c

@@ -39,7 +39,7 @@ str dmq_node_timeout_str = str_init("timeout");
 /**
  * @brief get the string status of the node
  */
-str *get_status_str(int status)
+str *dmq_get_status_str(int status)
 {
 	switch(status) {
 		case DMQ_NODE_ACTIVE: {
@@ -389,8 +389,8 @@ int build_node_str(dmq_node_t *node, char *buf, int buflen)
 	len += 1;
 	memcpy(buf + len, "status=", 7);
 	len += 7;
-	memcpy(buf + len, get_status_str(node->status)->s,
-			get_status_str(node->status)->len);
-	len += get_status_str(node->status)->len;
+	memcpy(buf + len, dmq_get_status_str(node->status)->s,
+			dmq_get_status_str(node->status)->len);
+	len += dmq_get_status_str(node->status)->len;
 	return len;
 }

src/modules/dmq/dmqnode.h

@@ -75,7 +75,7 @@ void shm_free_node(dmq_node_t *node);
 void pkg_free_node(dmq_node_t *node);
 int set_dmq_node_params(dmq_node_t *node, param_t *params);
 
-str *get_status_str(int status);
+str *dmq_get_status_str(int status);
 int build_node_str(dmq_node_t *node, char *buf, int buflen);
 
 extern dmq_node_t *self_node;

src/modules/presence/Makefile

@@ -22,4 +22,5 @@ DEFS+=-DKAMAILIO_MOD_INTERFACE
 
 SERLIBPATH=../../lib
 SER_LIBS+=$(SERLIBPATH)/srdb1/srdb1
+SER_LIBS+=$(SERLIBPATH)/srutils/srutils
 include ../../Makefile.modules

src/modules/presence/doc/presence_admin.xml

@@ -62,6 +62,11 @@
 				<emphasis>tm</emphasis>.
 			</para>
 			</listitem>
+			<listitem>
+			<para>
+				<emphasis>dmq (only if replication is enabled)</emphasis>.
+			</para>
+			</listitem>
 			</itemizedlist>
 		</para>
 	</section>
@@ -897,6 +902,33 @@ modparam("presence", "retrieve_order_by", "priority, received_time")
     </example>
 </section>
 
+<section id="presence.p.enable_dmq">
+	<title><varname>enable_dmq</varname> (integer)</title>
+	<para>
+		If set to 1, will enable DMQ replication of presentities between nodes. Use this instead of a shared DB
+		to share state across a cluster and update local watchers in realtime (subs_db_mode < 3) or on next
+		notifier run (subs_db_mode = 3).
+	</para>
+	<para>
+		<emphasis>
+			If this parameter is enabled, the DMQ module must be loaded first - otherwise, startup will fail.
+		</emphasis>
+	</para>
+	<para>
+		<emphasis>
+			Default value is 0.
+		</emphasis>
+	</para>
+	<example>
+		<title>Set <varname>enable_dmq</varname> parameter</title>
+		<programlisting format="linespecific">
+			...
+			modparam("presence", "enable_dmq", 1)
+			...
+		</programlisting>
+	</example>
+</section>
+
 </section>
 
 <section>

src/modules/presence/notify.c

@@ -79,6 +79,7 @@ str str_watcher_domain_col = str_init("watcher_domain");
 str str_event_id_col = str_init("event_id");
 str str_event_col = str_init("event");
 str str_etag_col = str_init("etag");
+str str_ruid_col = str_init("ruid");
 str str_from_tag_col = str_init("from_tag");
 str str_to_tag_col = str_init("to_tag");
 str str_callid_col = str_init("callid");

src/modules/presence/notify.h

@@ -76,6 +76,7 @@ extern str str_watcher_domain_col;
 extern str str_event_id_col;
 extern str str_event_col;
 extern str str_etag_col;
+extern str str_ruid_col;
 extern str str_from_tag_col;
 extern str str_to_tag_col;
 extern str str_callid_col;

src/modules/presence/presence.c

@@ -68,6 +68,7 @@
 #include "event_list.h"
 #include "bind_presence.h"
 #include "notify.h"
+#include "presence_dmq.h"
 #include "../../core/mod_fix.h"
 #include "../../core/kemi.h"
 #include "../../core/timer_proc.h"
@@ -165,6 +166,7 @@ int pres_startup_mode = 1;
 str pres_xavp_cfg = {0};
 int pres_retrieve_order = 0;
 str pres_retrieve_order_by = str_init("priority");
+int pres_enable_dmq = 0;
 
 int db_table_lock_type = 1;
 db_locking_t db_table_lock = DB_LOCKING_WRITE;
@@ -174,6 +176,8 @@ int *pres_notifier_id = NULL;
 int phtable_size= 9;
 phtable_t* pres_htable=NULL;
 
+sruid_t pres_sruid;
+
 static cmd_export_t cmds[]=
 {
 	{"handle_publish",        (cmd_function)w_handle_publish,        0,
@@ -233,7 +237,8 @@ static param_export_t params[]={
 	{ "retrieve_order",         PARAM_INT, &pres_retrieve_order},
 	{ "retrieve_order_by",      PARAM_STR, &pres_retrieve_order_by},
 	{ "sip_uri_match",          PARAM_INT, &pres_uri_match},
-    { "cseq_offset",            PARAM_INT, &pres_cseq_offset},
+	{ "cseq_offset",            PARAM_INT, &pres_cseq_offset},
+	{ "enable_dmq",             PARAM_INT, &pres_enable_dmq},
 	{0,0,0}
 };
 
@@ -293,6 +298,10 @@ static int mod_init(void)
 		return 0;
 	}
 
+	if(sruid_init(&pres_sruid, '-', "pres", SRUID_INC) < 0) {
+		return -1;
+	}
+
 	if(expires_offset<0)
 		expires_offset = 0;
 
@@ -463,6 +472,11 @@ static int mod_init(void)
 	if (goto_on_notify_reply>=0 && event_rt.rlist[goto_on_notify_reply]==0)
 		goto_on_notify_reply=-1; /* disable */
 
+	if (pres_enable_dmq>0 && pres_dmq_initialize()!=0) {
+		LM_ERR("failed to initialize dmq integration\n");
+		return -1;
+	}
+
 	return 0;
 }
 
@@ -479,6 +493,10 @@ static int child_init(int rank)
 	if(library_mode)
 		return 0;
 
+	if(sruid_init(&pres_sruid, '-', "pres", SRUID_INC) < 0) {
+		return -1;
+	}
+
 	if (rank == PROC_MAIN)
 	{
 		int i;

src/modules/presence/presence.h

@@ -34,6 +34,7 @@
 #include "../../modules/sl/sl.h"
 #include "../../lib/srdb1/db.h"
 #include "../../core/parser/parse_from.h"
+#include "../../lib/srutils/sruid.h"
 #include "event_list.h"
 #include "hash.h"
 
@@ -94,10 +95,13 @@ extern int pres_startup_mode;
 extern str pres_xavp_cfg;
 extern int pres_retrieve_order;
 extern str pres_retrieve_order_by;
+extern int pres_enable_dmq;
 
 extern int phtable_size;
 extern phtable_t* pres_htable;
 
+extern sruid_t pres_sruid;
+
 extern db_locking_t db_table_lock;
 
 int update_watchers_status(str pres_uri, pres_ev_t* ev, str* rules_doc);

src/modules/presence/presence_dmq.c

@@ -0,0 +1,489 @@
+/**
+*
+* Copyright (C) 2018 Charles Chance (Sipcentric Ltd)
+*
+* This file is part of Kamailio, a free SIP server.
+*
+* Kamailio is free software; you can redistribute it and/or modify
+* it under the terms of the GNU General Public License as published by
+* the Free Software Foundation; either version 2 of the License, or
+* (at your option) any later version
+*
+* Kamailio is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+* GNU General Public License for more details.
+*
+* You should have received a copy of the GNU General Public License
+* along with this program; if not, write to the Free Software
+* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+*
+*/
+
+#include "presence_dmq.h"
+
+static str pres_dmq_content_type = str_init("application/json");
+static str pres_dmq_200_rpl = str_init("OK");
+static str pres_dmq_400_rpl = str_init("Bad Request");
+static str pres_dmq_500_rpl = str_init("Server Internal Error");
+
+static int *pres_dmq_proc_init = 0;
+static int *pres_dmq_recv = 0;
+
+dmq_api_t pres_dmqb;
+dmq_peer_t *pres_dmq_peer = NULL;
+dmq_resp_cback_t pres_dmq_resp_callback = {&pres_dmq_resp_callback_f, 0};
+
+int pres_dmq_send_all_presentities();
+int pres_dmq_request_sync();
+
+/**
+* @brief add notification peer
+*/
+int pres_dmq_initialize()
+{
+	dmq_peer_t not_peer;
+
+	/* load the DMQ API */
+	if(dmq_load_api(&pres_dmqb) != 0) {
+		LM_ERR("cannot load dmq api\n");
+		return -1;
+	} else {
+		LM_DBG("loaded dmq api\n");
+	}
+
+	not_peer.callback = pres_dmq_handle_msg;
+	not_peer.init_callback = pres_dmq_request_sync;
+	not_peer.description.s = "presence";
+	not_peer.description.len = 8;
+	not_peer.peer_id.s = "presence";
+	not_peer.peer_id.len = 8;
+	pres_dmq_peer = pres_dmqb.register_dmq_peer(&not_peer);
+	if(!pres_dmq_peer) {
+		LM_ERR("error in register_dmq_peer\n");
+		goto error;
+	} else {
+		LM_DBG("dmq peer registered\n");
+	}
+	return 0;
+error:
+	return -1;
+}
+
+static int pres_dmq_init_proc()
+{
+	// TODO: tidy up
+
+	if(!pres_dmq_proc_init) {
+		LM_DBG("Initializing pres_dmq_proc_init for pid (%d)\n", my_pid());
+		pres_dmq_proc_init = (int *)pkg_malloc(sizeof(int));
+		if(!pres_dmq_proc_init) {
+			LM_ERR("no more pkg memory\n");
+			return -1;
+		}
+		*pres_dmq_proc_init = 0;
+	}
+
+	if(!pres_dmq_recv) {
+		LM_DBG("Initializing pres_dmq_recv for pid (%d)\n", my_pid());
+		pres_dmq_recv = (int *)pkg_malloc(sizeof(int));
+		if(!pres_dmq_recv) {
+			LM_ERR("no more pkg memory\n");
+			return -1;
+		}
+		*pres_dmq_recv = 0;
+	}
+
+	if(pres_sruid.pid == 0) {
+		LM_DBG("Initializing pres_sruid for pid (%d)\n", my_pid());
+		if(sruid_init(&pres_sruid, '-', "pres", SRUID_INC) < 0) {
+			return -1;
+		}
+	}
+
+	if(!pa_db) {
+		LM_DBG("Initializing presence DB connection for pid (%d)\n", my_pid());
+
+		if(pa_dbf.init == 0) {
+			LM_ERR("dmq_worker_init: database not bound\n");
+			return -1;
+		}
+
+		/* Do not pool the connections where possible when running notifier
+		* processes. */
+		if(pres_notifier_processes > 0 && pa_dbf.init2)
+			pa_db = pa_dbf.init2(&db_url, DB_POOLING_NONE);
+		else
+			pa_db = pa_dbf.init(&db_url);
+
+		if(!pa_db) {
+			LM_ERR("dmq_worker_init: unsuccessful database connection\n");
+			return -1;
+		}
+	}
+
+	*pres_dmq_proc_init = 1;
+
+	LM_DBG("process initialization complete\n");
+
+	return 0;
+}
+
+int pres_dmq_send(str *body, dmq_node_t *node)
+{
+	if(!pres_dmq_peer) {
+		LM_ERR("pres_dmq_peer is null!\n");
+		return -1;
+	}
+	if(node) {
+		LM_DBG("sending dmq message ...\n");
+		pres_dmqb.send_message(pres_dmq_peer, body, node,
+				&pres_dmq_resp_callback, 1, &pres_dmq_content_type);
+	} else {
+		LM_DBG("sending dmq broadcast...\n");
+		pres_dmqb.bcast_message(pres_dmq_peer, body, 0, &pres_dmq_resp_callback,
+				1, &pres_dmq_content_type);
+	}
+	return 0;
+}
+
+/**
+ * @brief extract presentity from json object
+*/
+presentity_t *pres_parse_json_presentity(srjson_t *in)
+{
+
+	int p_expires = 0, p_recv = 0;
+	str p_domain = STR_NULL, p_user = STR_NULL, p_etag = STR_NULL,
+		p_sender = STR_NULL, p_event_str = STR_NULL;
+	srjson_t *p_it;
+	pres_ev_t *p_event = NULL;
+	presentity_t *presentity = NULL;
+
+	LM_DBG("extracting presentity\n");
+
+	for(p_it = in->child; p_it; p_it = p_it->next) {
+		if(strcmp(p_it->string, "domain") == 0) {
+			p_domain.s = p_it->valuestring;
+			p_domain.len = strlen(p_it->valuestring);
+		} else if(strcmp(p_it->string, "user") == 0) {
+			p_user.s = p_it->valuestring;
+			p_user.len = strlen(p_it->valuestring);
+		} else if(strcmp(p_it->string, "etag") == 0) {
+			p_etag.s = p_it->valuestring;
+			p_etag.len = strlen(p_it->valuestring);
+		} else if(strcmp(p_it->string, "expires") == 0) {
+			p_expires = SRJSON_GET_INT(p_it);
+		} else if(strcmp(p_it->string, "recv") == 0) {
+			p_recv = SRJSON_GET_INT(p_it);
+		} else if(strcmp(p_it->string, "sender") == 0) {
+			p_sender.s = p_it->valuestring;
+			p_sender.len = strlen(p_it->valuestring);
+		} else if(strcmp(p_it->string, "event") == 0) {
+			p_event_str.s = p_it->valuestring;
+			p_event_str.len = strlen(p_it->valuestring);
+			p_event = contains_event(&p_event_str, 0);
+			if(!p_event) {
+				LM_ERR("unsupported event %s\n", p_it->valuestring);
+				return NULL;
+			}
+		} else {
+			LM_ERR("unrecognized field in json object\n");
+			return NULL;
+		}
+	}
+
+	LM_DBG("building presentity from domain: %.*s, user: %.*s, expires: %d, "
+		   "event: "
+		   "%.*s, etag: %.*s, sender: %.*s",
+			p_domain.len, p_domain.s, p_user.len, p_user.s, p_expires,
+			p_event->name.len, p_event->name.s, p_etag.len, p_etag.s,
+			p_sender.len, p_sender.s);
+
+	presentity = new_presentity(
+			&p_domain, &p_user, p_expires, p_event, &p_etag, &p_sender);
+
+	if(!presentity)
+		return NULL;
+
+	if(p_recv > 0)
+		presentity->received_time = p_recv;
+
+	return presentity;
+}
+
+/**
+* @brief presence dmq callback
+*/
+int pres_dmq_handle_msg(
+		struct sip_msg *msg, peer_reponse_t *resp, dmq_node_t *node)
+{
+	int content_length = 0, t_new = 0, sent_reply = 0;
+	str cur_etag = STR_NULL, body = STR_NULL, p_body = STR_NULL,
+		ruid = STR_NULL;
+	char *sphere = NULL;
+	srjson_doc_t jdoc;
+	srjson_t *it = NULL;
+	presentity_t *presentity = NULL;
+
+	pres_dmq_action_t action = PRES_DMQ_NONE;
+
+	/* received dmq message */
+	LM_DBG("dmq message received\n");
+
+	if(!pres_dmq_proc_init && pres_dmq_init_proc() < 0) {
+		return 0;
+	}
+
+	*pres_dmq_recv = 1;
+
+	if(!msg->content_length) {
+		LM_ERR("no content length header found\n");
+		goto invalid;
+	}
+	content_length = get_content_length(msg);
+	if(!content_length) {
+		LM_DBG("content length is 0\n");
+		goto invalid;
+	}
+
+	body.s = get_body(msg);
+	body.len = content_length;
+
+	if(!body.s) {
+		LM_ERR("unable to get body\n");
+		goto error;
+	}
+
+	/* parse body */
+	LM_DBG("body: %.*s\n", body.len, body.s);
+
+	srjson_InitDoc(&jdoc, NULL);
+	jdoc.buf = body;
+
+	if(jdoc.root == NULL) {
+		jdoc.root = srjson_Parse(&jdoc, jdoc.buf.s);
+		if(jdoc.root == NULL) {
+			LM_ERR("invalid json doc [[%s]]\n", jdoc.buf.s);
+			goto invalid;
+		}
+	}
+
+	/* iterate over keys */
+	for(it = jdoc.root->child; it; it = it->next) {
+		LM_DBG("found field: %s\n", it->string);
+		if(strcmp(it->string, "action") == 0) {
+			action = SRJSON_GET_INT(it);
+		} else if(strcmp(it->string, "presentity") == 0) {
+			presentity = pres_parse_json_presentity(it);
+			if(!presentity) {
+				LM_ERR("failed to construct presentity from json\n");
+				goto invalid;
+			}
+		} else if(strcmp(it->string, "t_new") == 0) {
+			t_new = SRJSON_GET_INT(it);
+		} else if(strcmp(it->string, "cur_etag") == 0) {
+			cur_etag.s = it->valuestring;
+			cur_etag.len = strlen(it->valuestring);
+		} else if(strcmp(it->string, "sphere") == 0) {
+			sphere = it->valuestring;
+		} else if(strcmp(it->string, "ruid") == 0) {
+			ruid.s = it->valuestring;
+			ruid.len = strlen(it->valuestring);
+		} else if(strcmp(it->string, "body") == 0) {
+			p_body.s = it->valuestring;
+			p_body.len = strlen(it->valuestring);
+		} else {
+			LM_ERR("unrecognized field in json object\n");
+			goto invalid;
+		}
+	}
+
+	switch(action) {
+		case PRES_DMQ_UPDATE_PRESENTITY:
+			if(update_presentity(NULL, presentity, &p_body, t_new, &sent_reply,
+					   sphere, &cur_etag, &ruid)
+					< 0) {
+				goto error;
+			}
+			break;
+		case PRES_DMQ_SYNC:
+		case PRES_DMQ_NONE:
+			break;
+	}
+
+	resp->reason = pres_dmq_200_rpl;
+	resp->resp_code = 200;
+	goto cleanup;
+
+invalid:
+	resp->reason = pres_dmq_400_rpl;
+	resp->resp_code = 400;
+	goto cleanup;
+
+error:
+	resp->reason = pres_dmq_500_rpl;
+	resp->resp_code = 500;
+
+cleanup:
+	*pres_dmq_recv = 0;
+	srjson_DestroyDoc(&jdoc);
+	if(presentity)
+		pkg_free(presentity);
+
+	return 0;
+}
+
+
+int pres_dmq_request_sync()
+{
+	srjson_doc_t jdoc;
+
+	LM_DBG("requesting sync from dmq peers\n");
+
+	srjson_InitDoc(&jdoc, NULL);
+
+	jdoc.root = srjson_CreateObject(&jdoc);
+	if(jdoc.root == NULL) {
+		LM_ERR("cannot create json root\n");
+		goto error;
+	}
+
+	srjson_AddNumberToObject(&jdoc, jdoc.root, "action", PRES_DMQ_SYNC);
+	jdoc.buf.s = srjson_PrintUnformatted(&jdoc, jdoc.root);
+	if(jdoc.buf.s == NULL) {
+		LM_ERR("unable to serialize data\n");
+		goto error;
+	}
+	jdoc.buf.len = strlen(jdoc.buf.s);
+	LM_DBG("sending serialized data %.*s\n", jdoc.buf.len, jdoc.buf.s);
+	if(pres_dmq_send(&jdoc.buf, 0) != 0) {
+		goto error;
+	}
+
+	jdoc.free_fn(jdoc.buf.s);
+	jdoc.buf.s = NULL;
+	srjson_DestroyDoc(&jdoc);
+	return 0;
+
+error:
+	if(jdoc.buf.s != NULL) {
+		jdoc.free_fn(jdoc.buf.s);
+		jdoc.buf.s = NULL;
+	}
+	srjson_DestroyDoc(&jdoc);
+	return -1;
+}
+
+
+int pres_dmq_replicate_presentity(presentity_t *presentity, str *body,
+		int t_new, str *cur_etag, char *sphere, str *ruid, dmq_node_t *node)
+{
+
+	srjson_doc_t jdoc;
+	srjson_t *p_json;
+
+	LM_DBG("replicating presentity record - old etag %.*s, new etag %.*s, ruid "
+		   "%.*s\n",
+			presentity->etag.len, presentity->etag.s, cur_etag->len,
+			cur_etag->s, ruid->len, ruid->s);
+
+	if(!pres_dmq_proc_init && pres_dmq_init_proc() < 0) {
+		return -1;
+	}
+
+	if(*pres_dmq_recv) {
+		return 0;
+	}
+
+	srjson_InitDoc(&jdoc, NULL);
+
+	jdoc.root = srjson_CreateObject(&jdoc);
+	if(jdoc.root == NULL) {
+		LM_ERR("cannot create json root\n");
+		goto error;
+	}
+
+	// action
+	srjson_AddNumberToObject(
+			&jdoc, jdoc.root, "action", PRES_DMQ_UPDATE_PRESENTITY);
+	// presentity
+	p_json = srjson_CreateObject(&jdoc);
+	srjson_AddStrToObject(&jdoc, p_json, "domain", presentity->domain.s,
+			presentity->domain.len);
+	srjson_AddStrToObject(
+			&jdoc, p_json, "user", presentity->user.s, presentity->user.len);
+	srjson_AddStrToObject(
+			&jdoc, p_json, "etag", presentity->etag.s, presentity->etag.len);
+	srjson_AddNumberToObject(&jdoc, p_json, "expires", presentity->expires);
+	srjson_AddNumberToObject(&jdoc, p_json, "recv", presentity->received_time);
+	if(presentity->sender) {
+		srjson_AddStrToObject(&jdoc, p_json, "sender", presentity->sender->s,
+				presentity->sender->len);
+	}
+	srjson_AddStrToObject(&jdoc, p_json, "event", presentity->event->name.s,
+			presentity->event->name.len);
+	srjson_AddItemToObject(&jdoc, jdoc.root, "presentity", p_json);
+	// t_new
+	srjson_AddNumberToObject(&jdoc, jdoc.root, "t_new", t_new);
+	// cur_etag
+	if(cur_etag) {
+		srjson_AddStrToObject(
+				&jdoc, jdoc.root, "cur_etag", cur_etag->s, cur_etag->len);
+	}
+	// sphere
+	if(sphere) {
+		srjson_AddStringToObject(&jdoc, jdoc.root, "sphere", sphere);
+	}
+	// ruid
+	if(ruid) {
+		srjson_AddStrToObject(&jdoc, jdoc.root, "ruid", ruid->s, ruid->len);
+	}
+	// body
+	if(body) {
+		srjson_AddStrToObject(&jdoc, jdoc.root, "body", body->s, body->len);
+	}
+
+	jdoc.buf.s = srjson_PrintUnformatted(&jdoc, jdoc.root);
+	if(jdoc.buf.s == NULL) {
+		LM_ERR("unable to serialize data\n");
+		goto error;
+	}
+	jdoc.buf.len = strlen(jdoc.buf.s);
+	LM_DBG("sending serialized data %.*s\n", jdoc.buf.len, jdoc.buf.s);
+	if(pres_dmq_send(&jdoc.buf, node) != 0) {
+		goto error;
+	}
+
+	jdoc.free_fn(jdoc.buf.s);
+	jdoc.buf.s = NULL;
+	srjson_DestroyDoc(&jdoc);
+	return 0;
+
+error:
+	if(jdoc.buf.s != NULL) {
+		jdoc.free_fn(jdoc.buf.s);
+		jdoc.buf.s = NULL;
+	}
+	srjson_DestroyDoc(&jdoc);
+	return -1;
+}
+
+
+int pres_dmq_send_all_presentities(dmq_node_t *dmq_node)
+{
+	// TODO: implement send all presentities
+
+	return 0;
+}
+
+
+/**
+* @brief dmq response callback
+*/
+int pres_dmq_resp_callback_f(
+		struct sip_msg *msg, int code, dmq_node_t *node, void *param)
+{
+	LM_DBG("dmq response callback triggered [%p %d %p]\n", msg, code, param);
+	return 0;
+}

src/modules/presence/presence_dmq.h

@@ -0,0 +1,50 @@
+/**
+ *
+ * Copyright (C) 2018 Charles Chance (Sipcentric Ltd)
+ *
+ * This file is part of Kamailio, a free SIP server.
+ *
+ * Kamailio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version
+ *
+ * Kamailio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License 
+ * along with this program; if not, write to the Free Software 
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef _PRESENCE_DMQ_H_
+#define _PRESENCE_DMQ_H_
+
+#include "presence.h"
+#include "presentity.h"
+#include "../dmq/bind_dmq.h"
+#include "../../lib/srutils/srjson.h"
+#include "../../core/strutils.h"
+#include "../../core/parser/msg_parser.h"
+#include "../../core/parser/parse_content.h"
+
+extern dmq_api_t pres_dmqb;
+extern dmq_peer_t* pres_dmq_peer;
+extern dmq_resp_cback_t pres_dmq_resp_callback;
+
+typedef enum {
+	PRES_DMQ_NONE,
+	PRES_DMQ_UPDATE_PRESENTITY,
+	PRES_DMQ_SYNC,
+} pres_dmq_action_t;
+
+int pres_dmq_initialize();
+int pres_dmq_handle_msg(struct sip_msg* msg, peer_reponse_t* resp,
+		dmq_node_t* node);
+int pres_dmq_replicate_presentity(presentity_t* presentity, str* body, int new_t, 
+		str* cur_etag, char* sphere, str* ruid, dmq_node_t* node);
+int pres_dmq_resp_callback_f(struct sip_msg* msg, int code, dmq_node_t* node,
+		void* param);
+#endif

src/modules/presence/presentity.c

@@ -44,6 +44,7 @@
 #include "publish.h"
 #include "hash.h"
 #include "utils_func.h"
+#include "presence_dmq.h"
 
 
 /* base priority value (20150101T000000) */
@@ -425,7 +426,7 @@ int delete_presentity_if_dialog_id_exists(presentity_t* presentity, char* dialog
 
 				LM_WARN("Presentity already exists - deleting it\n");
 
-				if(delete_presentity(&old_presentity)<0) {
+				if(delete_presentity(&old_presentity, NULL)<0) {
 					LM_ERR("failed to delete presentity\n");
 				}
 
@@ -555,20 +556,21 @@ int is_dialog_terminated(presentity_t* presentity)
 }
 
 int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
-		int new_t, int* sent_reply, char* sphere)
+		int new_t, int* sent_reply, char* sphere, str* etag_override, str* ruid)
 {
-	db_key_t query_cols[13], update_keys[9], result_cols[6];
-	db_op_t  query_ops[13];
-	db_val_t query_vals[13], update_vals[9];
+	db_key_t query_cols[14], rquery_cols[2], update_keys[9], result_cols[7];
+	db_op_t  query_ops[14], rquery_ops[2];
+	db_val_t query_vals[14], rquery_vals[2], update_vals[9];
 	db1_res_t *result= NULL;
 	int n_query_cols = 0;
+	int n_rquery_cols = 0;
 	int n_update_cols = 0;
 	char* dot= NULL;
 	str etag= {0, 0};
 	str cur_etag= {0, 0};
 	str* rules_doc= NULL;
 	str pres_uri= {0, 0};
-	int rez_body_col, rez_sender_col, n_result_cols= 0;
+	int rez_body_col, rez_sender_col, rez_ruid_col, n_result_cols= 0;
 	db_row_t *row = NULL ;
 	db_val_t *row_vals = NULL;
 	str old_body, sender;
@@ -578,6 +580,8 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 	int db_record_exists = 0;
 	int num_watchers = 0;
 	char *old_dialog_id = NULL, *dialog_id = NULL;
+	str cur_ruid= {0, 0};
+	str p_ruid = {0, 0};
 
 	if (sent_reply) *sent_reply= 0;
 	if(pres_notifier_processes == 0 && presentity->event->req_auth)
@@ -628,9 +632,22 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 
 	result_cols[rez_body_col= n_result_cols++] = &str_body_col;
 	result_cols[rez_sender_col= n_result_cols++] = &str_sender_col;
+	result_cols[rez_ruid_col= n_result_cols++] = &str_ruid_col;
 
 	if(new_t)
 	{
+		LM_DBG("new presentity with etag %.*s\n", presentity->etag.len, presentity->etag.s);
+
+		if (ruid) {
+			/* use the provided ruid */
+			p_ruid = *ruid;
+		} else {
+			/* generate a new ruid */
+			if(sruid_next(&pres_sruid)<0)
+				goto error;
+			p_ruid = pres_sruid.uid;
+		}
+
 		/* insert new record in hash_table */
 
 		if ( publ_cache_enabled &&
@@ -640,6 +657,8 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 			goto error;
 		}
 
+		LM_DBG("new htable record added\n");
+
 		/* insert new record into database */
 		query_cols[n_query_cols] = &str_sender_col;
 		query_vals[n_query_cols].type = DB1_STR;
@@ -672,6 +691,12 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 		query_vals[n_query_cols].val.int_val = presentity->priority;
 		n_query_cols++;
 
+		query_cols[n_query_cols] = &str_ruid_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = p_ruid;
+		n_query_cols++;
+
 		if (presentity->expires != -1)
 		{
 			/* A real PUBLISH */
@@ -765,6 +790,21 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 	else
 	{
 
+		LM_DBG("updating existing presentity with etag %.*s\n", presentity->etag.len, presentity->etag.s);
+
+		if (ruid) {
+			p_ruid = *ruid;
+
+			rquery_cols[n_rquery_cols] = &str_ruid_col;
+			rquery_ops[n_rquery_cols] = OP_EQ;
+			rquery_vals[n_rquery_cols].type = DB1_STR;
+			rquery_vals[n_rquery_cols].nul = 0;
+			rquery_vals[n_rquery_cols].val.str_val = p_ruid;
+			n_rquery_cols++;
+
+			// TODO: check for out-of-sequence updates
+		}
+
 		if (pa_dbf.use_table(pa_db, &presentity_table) < 0)
 		{
 			LM_ERR("unsuccessful sql use table\n");
@@ -783,8 +823,10 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 		if(EVENT_DIALOG_SLA(presentity->event->evp))
 		{
 
-			if (pa_dbf.query (pa_db, query_cols, query_ops, query_vals,
-					result_cols, n_query_cols, n_result_cols, 0, &result) < 0)
+			if (pa_dbf.query (pa_db, ruid?rquery_cols:query_cols, 
+					ruid?rquery_ops:query_ops, ruid?rquery_vals:query_vals,
+					result_cols, ruid?n_rquery_cols:n_query_cols, n_result_cols,
+					0, &result) < 0)
 			{
 				LM_ERR("unsuccessful sql query\n");
 				goto error;
@@ -800,6 +842,19 @@ int update_presentity(struct sip_msg* msg, presentity_t* presentity, str* body,
 			row = &result->rows[0];
 			row_vals = ROW_VALUES(row);
 
+			/* store current ruid if we don't already know it */
+			if (!p_ruid.s && row_vals[rez_ruid_col].val.string_val) {
+				cur_ruid.len = strlen((char *) row_vals[rez_ruid_col].val.string_val);
+				cur_ruid.s = (char *) pkg_malloc(sizeof(char) * cur_ruid.len);
+				if (!cur_ruid.s)
+				{
+					LM_ERR("no private memory\n");
+					goto error;
+				}
+				memcpy(cur_ruid.s, (char *) row_vals[rez_ruid_col].val.string_val, cur_ruid.len);
+				p_ruid = cur_ruid;
+			}
+
 			old_body.s = (char*)row_vals[rez_body_col].val.string_val;
 			old_body.len = strlen(old_body.s);
 			if(check_if_dialog(*body, &is_dialog, &dialog_id)< 0)
@@ -864,8 +919,10 @@ after_dialog_check:
 
 			if (!db_record_exists)
 			{
-				if (pa_dbf.query (pa_db, query_cols, query_ops, query_vals,
-					result_cols, n_query_cols, n_result_cols, 0, &result) < 0)
+				if (pa_dbf.query (pa_db, ruid?rquery_cols:query_cols, 
+						ruid?rquery_ops:query_ops, ruid?rquery_vals:query_vals,
+						result_cols, ruid?n_rquery_cols:n_query_cols, n_result_cols,
+						0, &result) < 0)
 				{
 					LM_ERR("unsuccessful sql query\n");
 					goto error;
@@ -878,6 +935,22 @@ after_dialog_check:
 
 				db_record_exists = 1;
 
+				row = &result->rows[0];
+				row_vals = ROW_VALUES(row);
+
+				/* store current ruid if we don't already know it */
+				if (!p_ruid.s && row_vals[rez_ruid_col].val.string_val) {
+					cur_ruid.len = strlen((char *) row_vals[rez_ruid_col].val.string_val);
+					cur_ruid.s = (char *) pkg_malloc(sizeof(char) * cur_ruid.len);
+					if (!cur_ruid.s)
+					{
+						LM_ERR("no private memory\n");
+						goto error;
+					}
+					memcpy(cur_ruid.s, (char *) row_vals[rez_ruid_col].val.string_val, cur_ruid.len);
+					p_ruid = cur_ruid;
+				}
+
 				pa_dbf.free_result(pa_db, result);
 				result = NULL;
 			}
@@ -899,7 +972,7 @@ after_dialog_check:
 
 				if (num_watchers > 0)
 				{
-					if (mark_presentity_for_delete(presentity) < 0)
+					if (mark_presentity_for_delete(presentity, &p_ruid) < 0)
 					{
 						LM_ERR("Marking presentities\n");
 						goto error;
@@ -917,7 +990,7 @@ after_dialog_check:
 
 			if (pres_notifier_processes == 0 || num_watchers == 0)
 			{
-				if (delete_presentity(presentity) < 0)
+				if (delete_presentity(presentity, &p_ruid) < 0)
 				{
 					LM_ERR("Deleting presentity\n");
 					goto error;
@@ -953,8 +1026,15 @@ after_dialog_check:
 			goto done;
 		}
 
-		if(presentity->event->etag_not_new== 0)
+		if(presentity->event->etag_not_new== 0 || etag_override)
 		{
+			if (etag_override) {
+				/* use the supplied etag */
+				LM_DBG("updating with supplied etag %.*s\n", etag_override->len, etag_override->s);
+				cur_etag = *etag_override;
+				goto after_etag_generation;
+			}
+
 			/* generate another etag */
 			unsigned int publ_nr;
 			str str_publ_nr= {0, 0};
@@ -988,10 +1068,12 @@ after_dialog_check:
 
 			cur_etag= etag;
 
+after_etag_generation:
+
 			update_keys[n_update_cols] = &str_etag_col;
 			update_vals[n_update_cols].type = DB1_STR;
 			update_vals[n_update_cols].nul = 0;
-			update_vals[n_update_cols].val.str_val = etag;
+			update_vals[n_update_cols].val.str_val = cur_etag;
 			n_update_cols++;
 
 		}
@@ -1068,11 +1150,14 @@ after_dialog_check:
 			n_update_cols++;
 		}
 
-		/* if there is no support for affected_rows and no previous query has been done, do query */
-		if (!pa_dbf.affected_rows && !db_record_exists)
+		/* if there is no support for affected_rows and no previous query has been done,
+		 * or dmq replication is enabled and we don't already know the ruid, do query */
+		if ((!pa_dbf.affected_rows && !db_record_exists) || (pres_enable_dmq > 0 && !p_ruid.s))
 		{
-			if (pa_dbf.query (pa_db, query_cols, query_ops, query_vals,
-					result_cols, n_query_cols, n_result_cols, 0, &result) < 0)
+			if (pa_dbf.query (pa_db, ruid?rquery_cols:query_cols, 
+					ruid?rquery_ops:query_ops, ruid?rquery_vals:query_vals,
+					result_cols, ruid?n_rquery_cols:n_query_cols, n_result_cols,
+					0, &result) < 0)
 			{
 				LM_ERR("unsuccessful sql query\n");
 				goto error;
@@ -1084,12 +1169,33 @@ after_dialog_check:
 				goto send_412;
 
 			db_record_exists = 1;
+			affected_rows = result->n;
+
+			row = &result->rows[0];
+			row_vals = ROW_VALUES(row);
+
+			/* store current ruid if we don't already know it */
+			if (!p_ruid.s && row_vals[rez_ruid_col].val.string_val) {
+				cur_ruid.len = strlen((char *) row_vals[rez_ruid_col].val.string_val);
+				cur_ruid.s = (char *) pkg_malloc(sizeof(char) * cur_ruid.len);
+				if (!cur_ruid.s)
+				{
+					LM_ERR("no private memory\n");
+					goto error;
+				}
+				memcpy(cur_ruid.s, (char *) row_vals[rez_ruid_col].val.string_val, cur_ruid.len);
+				p_ruid = cur_ruid;
+
+				LM_DBG("existing ruid %.*s\n", p_ruid.len, p_ruid.s);
+			}
+
 			pa_dbf.free_result(pa_db, result);
 			result = NULL;
 		}
 
-		if( pa_dbf.update( pa_db,query_cols, query_ops, query_vals,
-				update_keys, update_vals, n_query_cols, n_update_cols )<0)
+		if (pa_dbf.update (pa_db, ruid?rquery_cols:query_cols, 
+				ruid?rquery_ops:query_ops, ruid?rquery_vals:query_vals,
+				update_keys, update_vals, ruid?n_rquery_cols:n_query_cols, n_update_cols) < 0)
 		{
 			LM_ERR("updating published info in database\n");
 			goto error;
@@ -1108,7 +1214,7 @@ after_dialog_check:
 
 
 		/*if either affected_rows (if exists) or select query show that there is no line in database*/
-		if ((pa_dbf.affected_rows && !affected_rows) || (!pa_dbf.affected_rows && !db_record_exists))
+		if ((pa_dbf.affected_rows && !affected_rows && !db_record_exists) || (!pa_dbf.affected_rows && !db_record_exists))
 			goto send_412;
 
 		/* send 200OK */
@@ -1148,6 +1254,15 @@ send_notify:
 	}
 
 done:
+
+	if (pres_enable_dmq>0) {
+		pres_dmq_replicate_presentity(presentity, body, new_t, &cur_etag, sphere, &p_ruid, NULL);
+	}
+
+	if(cur_ruid.s)
+		pkg_free(cur_ruid.s);
+	cur_ruid.s= NULL;
+
 	if(rules_doc)
 	{
 		if(rules_doc->s)
@@ -1173,7 +1288,12 @@ done:
 
 send_412:
 
-	LM_ERR("No E_Tag match %*s\n", presentity->etag.len, presentity->etag.s);
+	if (!ruid) {
+		LM_ERR("No E_Tag match %*s\n", presentity->etag.len, presentity->etag.s);
+	} else {
+		LM_ERR("No ruid match %*s\n", ruid->len, ruid->s);
+	}
+	
 	if (msg != NULL)
 	{
 		if (slb.freply(msg, 412, &pu_412_rpl) < 0)
@@ -1198,6 +1318,8 @@ error:
 	if(pres_uri.s) {
 		pkg_free(pres_uri.s);
 	}
+	if(cur_ruid.s)
+		pkg_free(cur_ruid.s);
 
 	if (pa_dbf.abort_transaction) {
 		if (pa_dbf.abort_transaction(pa_db) < 0) {
@@ -1521,7 +1643,7 @@ error:
 
 }
 
-int mark_presentity_for_delete(presentity_t *pres)
+int mark_presentity_for_delete(presentity_t *pres, str *ruid)
 {
 	db_key_t query_cols[4], result_cols[1], update_cols[3];
 	db_val_t query_vals[4], update_vals[3], *value;
@@ -1535,7 +1657,7 @@ int mark_presentity_for_delete(presentity_t *pres)
 	if (pres->event->agg_nbody == NULL)
 	{
 		/* Nothing clever to do here... just delete */
-		if (delete_presentity(pres) < 0)
+		if (delete_presentity(pres, NULL) < 0)
 		{
 			LM_ERR("deleting presentity\n");
 			goto error;
@@ -1549,29 +1671,37 @@ int mark_presentity_for_delete(presentity_t *pres)
 		goto error;
 	}
 
-	query_cols[n_query_cols] = &str_username_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->user;
-	n_query_cols++;
+	if (!ruid) {
+		query_cols[n_query_cols] = &str_username_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->user;
+		n_query_cols++;
 
-	query_cols[n_query_cols] = &str_domain_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->domain;
-	n_query_cols++;
+		query_cols[n_query_cols] = &str_domain_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->domain;
+		n_query_cols++;
 
-	query_cols[n_query_cols] = &str_event_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->event->name;
-	n_query_cols++;
+		query_cols[n_query_cols] = &str_event_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->event->name;
+		n_query_cols++;
 
-	query_cols[n_query_cols] = &str_etag_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->etag;
-	n_query_cols++;
+		query_cols[n_query_cols] = &str_etag_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->etag;
+		n_query_cols++;
+	} else {
+		query_cols[n_query_cols] = &str_ruid_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = *ruid;
+		n_query_cols++;
+	}
 
 	result_cols[0] = &str_body_col;
 
@@ -1601,7 +1731,7 @@ int mark_presentity_for_delete(presentity_t *pres)
 		 * it anyway */
 		LM_ERR("Found %d presentities - expected 1\n", RES_ROW_N(result));
 
-		if (delete_presentity(pres) < 0)
+		if (delete_presentity(pres, ruid) < 0)
 		{
 			LM_ERR("deleting presentity\n");
 			goto error;
@@ -1669,7 +1799,7 @@ error:
 	return ret;
 }
 
-int delete_presentity(presentity_t *pres)
+int delete_presentity(presentity_t *pres, str *ruid)
 {
 	db_key_t query_cols[4];
 	db_val_t query_vals[4];
@@ -1681,29 +1811,37 @@ int delete_presentity(presentity_t *pres)
 		goto error;
 	}
 
-	query_cols[n_query_cols] = &str_username_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->user;
-	n_query_cols++;
+	if (!ruid) {
+		query_cols[n_query_cols] = &str_username_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->user;
+		n_query_cols++;
 
-	query_cols[n_query_cols] = &str_domain_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->domain;
-	n_query_cols++;
+		query_cols[n_query_cols] = &str_domain_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->domain;
+		n_query_cols++;
 
-	query_cols[n_query_cols] = &str_event_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->event->name;
-	n_query_cols++;
+		query_cols[n_query_cols] = &str_event_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->event->name;
+		n_query_cols++;
 
-	query_cols[n_query_cols] = &str_etag_col;
-	query_vals[n_query_cols].type = DB1_STR;
-	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.str_val = pres->etag;
-	n_query_cols++;
+		query_cols[n_query_cols] = &str_etag_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = pres->etag;
+		n_query_cols++;
+	} else {
+		query_cols[n_query_cols] = &str_ruid_col;
+		query_vals[n_query_cols].type = DB1_STR;
+		query_vals[n_query_cols].nul = 0;
+		query_vals[n_query_cols].val.str_val = *ruid;
+		n_query_cols++;		
+	}
 
 	if(pa_dbf.delete(pa_db, query_cols, 0, query_vals, n_query_cols) < 0)
 	{

src/modules/presence/presentity.h

@@ -32,7 +32,7 @@
 #include "../../core/str.h"
 #include "../../core/parser/msg_parser.h" 
 #include "event_list.h"
-//#include "presence.h"
+#include "presence.h"
 
 extern char prefix;
 
@@ -55,7 +55,7 @@ presentity_t* new_presentity( str* domain,str* user,int expires,
 
 /* update presentity in database */
 int update_presentity(struct sip_msg* msg,presentity_t* p,str* body,int t_new,
-		int* sent_reply, char* sphere);
+		int* sent_reply, char* sphere, str* etag_override, str* ruid);
 
 /* free memory */
 void free_presentity(presentity_t* p);
@@ -69,8 +69,8 @@ char* extract_sphere(str body);
 char* get_sphere(str* pres_uri);
 typedef char* (*pres_get_sphere_t)(str* pres_uri);
 
-int mark_presentity_for_delete(presentity_t *pres);
-int delete_presentity(presentity_t *pres);
+int mark_presentity_for_delete(presentity_t *pres, str *ruid);
+int delete_presentity(presentity_t *pres, str *ruid);
 int delete_offline_presentities(str *pres_uri, pres_ev_t *event);
 
 #endif

src/modules/presence/publish.c

@@ -157,7 +157,7 @@ void msg_presentity_clean(unsigned int ticks,void *param)
 
 			if (pres_force_delete == 1)
 			{
-				if (delete_presentity(&pres) < 0)
+				if (delete_presentity(&pres, NULL) < 0)
 				{
 					LM_ERR("Deleting presentity\n");
 					goto error;
@@ -186,7 +186,7 @@ void msg_presentity_clean(unsigned int ticks,void *param)
 
 				if (num_watchers > 0)
 				{
-					if (mark_presentity_for_delete(&pres) < 0)
+					if (mark_presentity_for_delete(&pres, NULL) < 0)
 					{
 						LM_ERR("Marking presentity\n");
 						if (pa_dbf.abort_transaction)
@@ -199,7 +199,7 @@ void msg_presentity_clean(unsigned int ticks,void *param)
 				}
 				else
 				{
-					if (delete_presentity(&pres) < 0)
+					if (delete_presentity(&pres, NULL) < 0)
 					{
 						LM_ERR("Deleting presentity\n");
 						goto error;
@@ -494,7 +494,7 @@ int ki_handle_publish_uri(struct sip_msg* msg, str* sender_uri)
 	}
 
 	/* querry the database and update or insert */
-	if(update_presentity(msg, presentity, &body, etag_gen, &sent_reply, sphere) <0)
+	if(update_presentity(msg, presentity, &body, etag_gen, &sent_reply, sphere, NULL, NULL) <0)
 	{
 		LM_ERR("when updating presentity\n");
 		goto error;
@@ -635,7 +635,7 @@ int update_hard_presentity(str *pres_uri, pres_ev_t *event, str *file_uri, str *
 		goto done;
 	}
 
-	if (update_presentity(NULL, pres, pidf_doc, new_t, NULL, sphere) < 0)
+	if (update_presentity(NULL, pres, pidf_doc, new_t, NULL, sphere, NULL, NULL) < 0)
 	{
 		LM_ERR("updating presentity\n");
 		goto done;

utils/kamctl/db_berkeley/kamailio/presentity

@@ -1,5 +1,5 @@
 METADATA_COLUMNS
-id(int) username(str) domain(str) event(str) etag(str) expires(int) received_time(int) body(str) sender(str) priority(int)
+id(int) username(str) domain(str) event(str) etag(str) expires(int) received_time(int) body(str) sender(str) priority(int) ruid(str)
 METADATA_KEY
 1 2 3 
 METADATA_READONLY
@@ -7,4 +7,4 @@ METADATA_READONLY
 METADATA_LOGFLAGS
 0
 METADATA_DEFAULTS
-NIL|NIL|NIL|NIL|NIL|NIL|NIL|NIL|NIL|0
+NIL|NIL|NIL|NIL|NIL|NIL|NIL|NIL|NIL|0|NIL

utils/kamctl/db_berkeley/kamailio/version

@@ -91,7 +91,7 @@ pdt|1
 pl_pipes|
 pl_pipes|1
 presentity|
-presentity|4
+presentity|5
 pua|
 pua|7
 purplemap|

utils/kamctl/db_sqlite/presence-create.sql

@@ -9,13 +9,15 @@ CREATE TABLE presentity (
     body BLOB NOT NULL,
     sender VARCHAR(128) NOT NULL,
     priority INTEGER DEFAULT 0 NOT NULL,
-    CONSTRAINT presentity_presentity_idx UNIQUE (username, domain, event, etag)
+    ruid VARCHAR(64),
+    CONSTRAINT presentity_presentity_idx UNIQUE (username, domain, event, etag),
+    CONSTRAINT presentity_ruid_idx UNIQUE (ruid)
 );
 
 CREATE INDEX presentity_presentity_expires ON presentity (expires);
 CREATE INDEX presentity_account_idx ON presentity (username, domain, event);
 
-INSERT INTO version (table_name, table_version) values ('presentity','4');
+INSERT INTO version (table_name, table_version) values ('presentity','5');
 
 CREATE TABLE active_watchers (
     id INTEGER PRIMARY KEY NOT NULL,

utils/kamctl/dbtext/kamailio/presentity

@@ -1 +1 @@
-id(int,auto) username(string) domain(string) event(string) etag(string) expires(int) received_time(int) body(string) sender(string) priority(int) 
+id(int,auto) username(string) domain(string) event(string) etag(string) expires(int) received_time(int) body(string) sender(string) priority(int) ruid(string,null) 

utils/kamctl/dbtext/kamailio/version

@@ -40,7 +40,7 @@ mtree:1
 mtrees:2
 pdt:1
 pl_pipes:1
-presentity:4
+presentity:5
 pua:7
 purplemap:1
 re_grp:1

utils/kamctl/mongodb/kamailio/presentity.json

@@ -1,6 +1,6 @@
 {
   "name": "presentity",
-  "version": 4,
+  "version": 5,
   "columns": [
     "id": {
       "type": "int",
@@ -51,6 +51,11 @@
       "type": "int",
       "default": 0,
       "null": false
+    },
+    "ruid": {
+      "type": "string",
+      "default": null,
+      "null": true
     }
   ]
 }

utils/kamctl/mongodb/kamailio/version-create.mongo

@@ -40,7 +40,7 @@ db.getCollection("version").insert({ table_name: "mtree", table_version: NumberI
 db.getCollection("version").insert({ table_name: "mtrees", table_version: NumberInt(2) });
 db.getCollection("version").insert({ table_name: "pdt", table_version: NumberInt(1) });
 db.getCollection("version").insert({ table_name: "pl_pipes", table_version: NumberInt(1) });
-db.getCollection("version").insert({ table_name: "presentity", table_version: NumberInt(4) });
+db.getCollection("version").insert({ table_name: "presentity", table_version: NumberInt(5) });
 db.getCollection("version").insert({ table_name: "pua", table_version: NumberInt(7) });
 db.getCollection("version").insert({ table_name: "purplemap", table_version: NumberInt(1) });
 db.getCollection("version").insert({ table_name: "re_grp", table_version: NumberInt(1) });

utils/kamctl/mysql/presence-create.sql

@@ -9,13 +9,15 @@ CREATE TABLE `presentity` (
     `body` BLOB NOT NULL,
     `sender` VARCHAR(128) NOT NULL,
     `priority` INT(11) DEFAULT 0 NOT NULL,
-    CONSTRAINT presentity_idx UNIQUE (`username`, `domain`, `event`, `etag`)
+    `ruid` VARCHAR(64),
+    CONSTRAINT presentity_idx UNIQUE (`username`, `domain`, `event`, `etag`),
+    CONSTRAINT ruid_idx UNIQUE (`ruid`)
 );
 
 CREATE INDEX presentity_expires ON presentity (`expires`);
 CREATE INDEX account_idx ON presentity (`username`, `domain`, `event`);
 
-INSERT INTO version (table_name, table_version) values ('presentity','4');
+INSERT INTO version (table_name, table_version) values ('presentity','5');
 
 CREATE TABLE `active_watchers` (
     `id` INT(10) UNSIGNED AUTO_INCREMENT PRIMARY KEY NOT NULL,

utils/kamctl/oracle/presence-create.sql

@@ -9,7 +9,9 @@ CREATE TABLE presentity (
     body BLOB,
     sender VARCHAR2(128),
     priority NUMBER(10) DEFAULT 0 NOT NULL,
-    CONSTRAINT presentity_presentity_idx  UNIQUE (username, domain, event, etag)
+    ruid VARCHAR2(64),
+    CONSTRAINT presentity_presentity_idx  UNIQUE (username, domain, event, etag),
+    CONSTRAINT presentity_ruid_idx  UNIQUE (ruid)
 );
 
 CREATE OR REPLACE TRIGGER presentity_tr
@@ -23,7 +25,7 @@ BEGIN map2users('presentity'); END;
 CREATE INDEX presentity_presentity_expires  ON presentity (expires);
 CREATE INDEX presentity_account_idx  ON presentity (username, domain, event);
 
-INSERT INTO version (table_name, table_version) values ('presentity','4');
+INSERT INTO version (table_name, table_version) values ('presentity','5');
 
 CREATE TABLE active_watchers (
     id NUMBER(10) PRIMARY KEY,

utils/kamctl/postgres/presence-create.sql

@@ -9,13 +9,15 @@ CREATE TABLE presentity (
     body BYTEA NOT NULL,
     sender VARCHAR(128) NOT NULL,
     priority INTEGER DEFAULT 0 NOT NULL,
-    CONSTRAINT presentity_presentity_idx UNIQUE (username, domain, event, etag)
+    ruid VARCHAR(64),
+    CONSTRAINT presentity_presentity_idx UNIQUE (username, domain, event, etag),
+    CONSTRAINT presentity_ruid_idx UNIQUE (ruid)
 );
 
 CREATE INDEX presentity_presentity_expires ON presentity (expires);
 CREATE INDEX presentity_account_idx ON presentity (username, domain, event);
 
-INSERT INTO version (table_name, table_version) values ('presentity','4');
+INSERT INTO version (table_name, table_version) values ('presentity','5');
 
 CREATE TABLE active_watchers (
     id SERIAL PRIMARY KEY NOT NULL,

utils/kamctl/xhttp_pi/pi_framework.xml

@@ -507,6 +507,7 @@
 		<column><field>body</field><type>DB1_BLOB</type></column>
 		<column><field>sender</field><type>DB1_STR</type></column>
 		<column><field>priority</field><type>DB1_INT</type></column>
+		<column><field>ruid</field><type>DB1_STR</type></column>
 	</db_table>
 	<!-- Declaration of active_watchers table-->
 	<db_table id="active_watchers">
@@ -2970,6 +2971,7 @@
 				<col><field>body</field></col>
 				<col><field>sender</field></col>
 				<col><field>priority</field></col>
+				<col><field>ruid</field></col>
 			</query_cols>
 		</cmd>
 		<cmd><cmd_name>add</cmd_name>
@@ -2985,6 +2987,7 @@
 				<col><field>body</field></col>
 				<col><field>sender</field></col>
 				<col><field>priority</field></col>
+				<col><field>ruid</field></col>
 			</query_cols>
 		</cmd>
 		<cmd><cmd_name>update</cmd_name>
@@ -3003,6 +3006,7 @@
 				<col><field>body</field></col>
 				<col><field>sender</field></col>
 				<col><field>priority</field></col>
+				<col><field>ruid</field></col>
 			</query_cols>
 		</cmd>
 		<cmd><cmd_name>delete</cmd_name>

utils/kamctl/xhttp_pi/presence-mod

@@ -14,6 +14,7 @@
 				<col><field>body</field></col>
 				<col><field>sender</field></col>
 				<col><field>priority</field></col>
+				<col><field>ruid</field></col>
 			</query_cols>
 		</cmd>
 		<cmd><cmd_name>add</cmd_name>
@@ -29,6 +30,7 @@
 				<col><field>body</field></col>
 				<col><field>sender</field></col>
 				<col><field>priority</field></col>
+				<col><field>ruid</field></col>
 			</query_cols>
 		</cmd>
 		<cmd><cmd_name>update</cmd_name>
@@ -47,6 +49,7 @@
 				<col><field>body</field></col>
 				<col><field>sender</field></col>
 				<col><field>priority</field></col>
+				<col><field>ruid</field></col>
 			</query_cols>
 		</cmd>
 		<cmd><cmd_name>delete</cmd_name>

utils/kamctl/xhttp_pi/presence-table

@@ -12,6 +12,7 @@
 		<column><field>body</field><type>DB1_BLOB</type></column>
 		<column><field>sender</field><type>DB1_STR</type></column>
 		<column><field>priority</field><type>DB1_INT</type></column>
+		<column><field>ruid</field><type>DB1_STR</type></column>
 	</db_table>
 	<!-- Declaration of active_watchers table-->
 	<db_table id="active_watchers">