@@ -410,6 +410,10 @@ error:

 	return -1;

 }

+/* Replay DMQ action

+

+Return 0 for non-error. Allt other returns are parsed as error.

+*/

 int ht_dmq_replay_action(ht_dmq_action_t action, str* htname, str* cname,

 		int type, int_str* val, int mode) {

@@ -350,7 +350,9 @@ error:

 	return 0;

 }

-int ht_dmq_replicate_action(ht_dmq_action_t action, str* htname, str* cname, int type, int_str* val, int mode) {

+int ht_dmq_replicate_action(ht_dmq_action_t action, str* htname, str* cname,

+		int type, int_str* val, int mode)

+{

 	srjson_doc_t jdoc;

@@ -370,7 +372,8 @@ int ht_dmq_replicate_action(ht_dmq_action_t action, str* htname, str* cname, int

 		srjson_AddStrToObject(&jdoc, jdoc.root, "cname", cname->s, cname->len);

 	}

-	if (action==HT_DMQ_SET_CELL || action==HT_DMQ_SET_CELL_EXPIRE || action==HT_DMQ_RM_CELL_RE) {

+	if (action==HT_DMQ_SET_CELL || action==HT_DMQ_SET_CELL_EXPIRE

+			|| action==HT_DMQ_RM_CELL_RE || action==HT_DMQ_RM_CELL_SW) {

 		srjson_AddNumberToObject(&jdoc, jdoc.root, "type", type);

 		if (type&AVP_VAL_STR) {

 			srjson_AddStrToObject(&jdoc, jdoc.root, "strval", val->s.s, val->s.len);

@@ -407,7 +410,8 @@ error:

 	return -1;

 }

-int ht_dmq_replay_action(ht_dmq_action_t action, str* htname, str* cname, int type, int_str* val, int mode) {

+int ht_dmq_replay_action(ht_dmq_action_t action, str* htname, str* cname,

+		int type, int_str* val, int mode) {

 	ht_t* ht;

 	ht = ht_get_table(htname);

@@ -416,7 +420,8 @@ int ht_dmq_replay_action(ht_dmq_action_t action, str* htname, str* cname, int ty

 		return -1;

 	}

-	LM_DBG("replaying action %d on %.*s=>%.*s...\n", action, htname->len, htname->s, cname->len, cname->s);

+	LM_DBG("replaying action %d on %.*s=>%.*s...\n", action,

+			htname->len, htname->s, cname->len, cname->s);

 	if (action==HT_DMQ_SET_CELL) {

 		return ht_set_cell(ht, cname, type, val, mode);

@@ -426,6 +431,8 @@ int ht_dmq_replay_action(ht_dmq_action_t action, str* htname, str* cname, int ty

 		return ht_del_cell(ht, cname);

 	} else if (action==HT_DMQ_RM_CELL_RE) {

 		return ht_rm_cell_re(&val->s, ht, mode);

+	} else if (action==HT_DMQ_RM_CELL_SW) {

+		return ht_rm_cell_op(&val->s, ht, mode, HT_RM_OP_SW);

 	} else {

 		LM_ERR("unrecognized action\n");

 		return -1;

@@ -48,7 +48,7 @@ static str dmq_500_rpl  = str_init("Server Internal Error");

 static int dmq_cell_group_empty_size = 12; // {"cells":[]}

 static int dmq_cell_group_max_size = 60000;

 static ht_dmq_jdoc_cell_group_t ht_dmq_jdoc_cell_group;

-int ht_dmq_init_sync;

+extern int ht_dmq_init_sync;

 dmq_api_t ht_dmqb;

 dmq_peer_t* ht_dmq_peer = NULL;

@@ -588,11 +588,11 @@ int ht_dmq_handle_sync(srjson_doc_t* jdoc) {

 			if(ht==NULL) {

 				LM_WARN("unable to get table %.*s\n",

 						htname.len, (htname.s)?htname.s:"");

-			}

-

-			if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0) {

-				LM_WARN("unable to set cell %.*s in table %.*s\n",

-						cname.len, cname.s, ht->name.len, ht->name.s);

+			} else {

+				if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0) {

+					LM_WARN("unable to set cell %.*s in table %.*s\n",

+							cname.len, cname.s, ht->name.len, ht->name.s);

+				}

 			}

 		}

@@ -139,36 +139,43 @@ static int ht_dmq_cell_group_flush(dmq_node_t* node) {

 	srjson_doc_t *jdoc = &ht_dmq_jdoc_cell_group.jdoc;

 	srjson_t *jdoc_cells = ht_dmq_jdoc_cell_group.jdoc_cells;

+	int ret = 0;

 	srjson_AddItemToObject(jdoc, jdoc->root, "cells", jdoc_cells);

-	LM_DBG("json[%s]\n", srjson_PrintUnformatted(jdoc, jdoc->root));

+	LM_DBG("jdoc size[%d]\n", ht_dmq_jdoc_cell_group.size);

 	jdoc->buf.s = srjson_PrintUnformatted(jdoc, jdoc->root);

 	if(jdoc->buf.s==NULL) {

 		LM_ERR("unable to serialize data\n");

-		return -1;

+		ret = -1;

+		goto cleanup;

 	}

 	jdoc->buf.len = strlen(jdoc->buf.s);

 	LM_DBG("sending serialized data %.*s\n", jdoc->buf.len, jdoc->buf.s);

 	if (ht_dmq_send(&jdoc->buf, node)!=0) {

 		LM_ERR("unable to send data\n");

-		return -1;

+		ret = -1;

 	}

-	LM_DBG("jdoc size[%d]\n", ht_dmq_jdoc_cell_group.size);

+cleanup:

+

+	srjson_DeleteItemFromObject(jdoc, jdoc->root, "cells");

+	ht_dmq_jdoc_cell_group.count = 0;

+	ht_dmq_jdoc_cell_group.size = dmq_cell_group_empty_size;

+

+	if(jdoc->buf.s!=NULL) {

+		jdoc->free_fn(jdoc->buf.s);

+		jdoc->buf.s = NULL;

+	}

-	srjson_Delete(jdoc, jdoc_cells);

 	ht_dmq_jdoc_cell_group.jdoc_cells = srjson_CreateArray(&ht_dmq_jdoc_cell_group.jdoc);

 	if (ht_dmq_jdoc_cell_group.jdoc_cells==NULL) {

 		LM_ERR("cannot re-create json cells array! \n");

-		return -1;

+		ret = -1;

 	}

-	ht_dmq_jdoc_cell_group.count = 0;

-	ht_dmq_jdoc_cell_group.size = dmq_cell_group_empty_size;

-

-	return 0;

+	return ret;

 }

 static void ht_dmq_cell_group_destroy() {

@@ -575,15 +575,18 @@ int ht_dmq_handle_sync(srjson_doc_t* jdoc) {

 			}

 		}

-		ht = ht_get_table(&htname);

-		if(ht==NULL) {

-			LM_WARN("unable to get table %.*s\n",

-					htname.len, (htname.s)?htname.s:"");

-		}

+		if(htname.s!=NULL && htname.len>0 && cname.s!=NULL

+				&& cname.len>0) {

+			ht = ht_get_table(&htname);

+			if(ht==NULL) {

+				LM_WARN("unable to get table %.*s\n",

+						htname.len, (htname.s)?htname.s:"");

+			}

-		if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0) {

-			LM_WARN("unable to set cell %.*s in table %.*s\n",

-					cname.len, cname.s, ht->name.len, ht->name.s);

+			if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0) {

+				LM_WARN("unable to set cell %.*s in table %.*s\n",

+						cname.len, cname.s, ht->name.len, ht->name.s);

+			}

 		}

 		cell = cell->next;

@@ -576,11 +576,15 @@ int ht_dmq_handle_sync(srjson_doc_t* jdoc) {

 		}

 		ht = ht_get_table(&htname);

-		if(ht==NULL)

-			LM_WARN("unable to get table %.*s\n", ht->name.len, ht->name.s);

+		if(ht==NULL) {

+			LM_WARN("unable to get table %.*s\n",

+					htname.len, (htname.s)?htname.s:"");

+		}

-		if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0)

-			LM_WARN("unable to set cell %.*s in table %.*s\n", cname.len, cname.s, ht->name.len, ht->name.s);

+		if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0) {

+			LM_WARN("unable to set cell %.*s in table %.*s\n",

+					cname.len, cname.s, ht->name.len, ht->name.s);

+		}

 		cell = cell->next;

 	}

@@ -537,17 +537,16 @@ error:

 int ht_dmq_handle_sync(srjson_doc_t* jdoc) {

 	LM_DBG("handling sync\n");

-	srjson_t* cells;

-	srjson_t* cell;

-	srjson_t* it;

-	str htname;

-	str cname;

-	int type;

-	int_str val;

-	int expire;

-	ht_t* ht;

-	time_t now;

-

+	srjson_t* cells = NULL;

+	srjson_t* cell = NULL;

+	srjson_t* it = NULL;

+	str htname = STR_NULL;

+	str cname = STR_NULL;

+	int type = 0;

+	int_str val = {0};

+	int expire = 0;

+	ht_t* ht = NULL;

+	time_t now = 0;

 	cells = jdoc->root->child;

 	cell = cells->child;

@@ -24,11 +24,6 @@

 #include "ht_dmq.h"

 #include "ht_api.h"

-static str ht_dmq_content_type = str_init("application/json");

-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");

-

 typedef struct _ht_dmq_repdata {

 	int action;

 	str htname;

@@ -39,10 +34,155 @@ typedef struct _ht_dmq_repdata {

 	int expire;

 } ht_dmq_repdata_t;

+typedef struct _ht_dmq_jdoc_cell_group {

+	int count;

+	int size;

+	srjson_doc_t jdoc;

+	srjson_t *jdoc_cells;

+} ht_dmq_jdoc_cell_group_t;

+

+static str ht_dmq_content_type = str_init("application/json");

+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 int dmq_cell_group_empty_size = 12; // {"cells":[]}

+static int dmq_cell_group_max_size = 60000;

+static ht_dmq_jdoc_cell_group_t ht_dmq_jdoc_cell_group;

+int ht_dmq_init_sync;

+

 dmq_api_t ht_dmqb;

 dmq_peer_t* ht_dmq_peer = NULL;

 dmq_resp_cback_t ht_dmq_resp_callback = {&ht_dmq_resp_callback_f, 0};

+int ht_dmq_send(str* body, dmq_node_t* node);

+int ht_dmq_send_sync(dmq_node_t* node);

+int ht_dmq_handle_sync(srjson_doc_t* jdoc);

+

+static int ht_dmq_cell_group_init(void) {

+

+	if (ht_dmq_jdoc_cell_group.jdoc.root)

+		return 0; // already initialised

+

+	ht_dmq_jdoc_cell_group.count = 0;

+	ht_dmq_jdoc_cell_group.size = dmq_cell_group_empty_size;

+

+	srjson_InitDoc(&ht_dmq_jdoc_cell_group.jdoc, NULL);

+

+	ht_dmq_jdoc_cell_group.jdoc.root = srjson_CreateObject(&ht_dmq_jdoc_cell_group.jdoc);

+	if (ht_dmq_jdoc_cell_group.jdoc.root==NULL) {

+		LM_ERR("cannot create json root object! \n");

+		return -1;

+	}

+

+	ht_dmq_jdoc_cell_group.jdoc_cells = srjson_CreateArray(&ht_dmq_jdoc_cell_group.jdoc);

+	if (ht_dmq_jdoc_cell_group.jdoc_cells==NULL) {

+		LM_ERR("cannot create json cells array! \n");

+		srjson_DestroyDoc(&ht_dmq_jdoc_cell_group.jdoc);

+		return -1;

+	}

+

+	return 0;

+}

+

+static int ht_dmq_cell_group_write(str* htname, ht_cell_t* ptr) {

+

+	// jsonify cell and add to array

+

+	str tmp;

+	srjson_doc_t *jdoc = &ht_dmq_jdoc_cell_group.jdoc;

+	srjson_t *jdoc_cells = ht_dmq_jdoc_cell_group.jdoc_cells;

+	srjson_t * jdoc_cell = srjson_CreateObject(jdoc);

+

+	if(!jdoc_cell) {

+		LM_ERR("cannot create cell json root\n");

+		return -1;

+	}

+

+	// add json overhead

+	if(ptr->flags&AVP_VAL_STR) {

+		ht_dmq_jdoc_cell_group.size += 54; // {"htname":"","cname":"","type":,"strval":"","expire":}

+	} else {

+		ht_dmq_jdoc_cell_group.size += 52; // {"htname":"","cname":"","type":,"intval":,"expire":}

+	}

+

+	srjson_AddStrToObject(jdoc, jdoc_cell, "htname", htname->s, htname->len);

+	ht_dmq_jdoc_cell_group.size += htname->len;

+

+	srjson_AddStrToObject(jdoc, jdoc_cell, "cname", ptr->name.s, ptr->name.len);

+	ht_dmq_jdoc_cell_group.size += ptr->name.len;

+

+	if (ptr->flags&AVP_VAL_STR) {

+		srjson_AddNumberToObject(jdoc, jdoc_cell, "type", AVP_VAL_STR);

+		ht_dmq_jdoc_cell_group.size += 1;

+		srjson_AddStrToObject(jdoc, jdoc_cell, "strval", ptr->value.s.s, ptr->value.s.len);

+		ht_dmq_jdoc_cell_group.size += ptr->value.s.len;

+	} else {

+		srjson_AddNumberToObject(jdoc, jdoc_cell, "type", 0);

+		ht_dmq_jdoc_cell_group.size += 1;

+		srjson_AddNumberToObject(jdoc, jdoc_cell, "intval", ptr->value.n);

+		tmp.s = sint2str((long)ptr->value.n, &tmp.len);

+		ht_dmq_jdoc_cell_group.size += tmp.len;

+	}

+

+	srjson_AddNumberToObject(jdoc, jdoc_cell, "expire", ptr->expire);

+	tmp.s = sint2str((long)ptr->expire, &tmp.len);

+	ht_dmq_jdoc_cell_group.size += tmp.len;

+

+	srjson_AddItemToArray(jdoc, jdoc_cells, jdoc_cell);

+

+	ht_dmq_jdoc_cell_group.count++;

+

+	return 0;

+}

+

+static int ht_dmq_cell_group_flush(dmq_node_t* node) {

+

+	srjson_doc_t *jdoc = &ht_dmq_jdoc_cell_group.jdoc;

+	srjson_t *jdoc_cells = ht_dmq_jdoc_cell_group.jdoc_cells;

+

+	srjson_AddItemToObject(jdoc, jdoc->root, "cells", jdoc_cells);

+

+	LM_DBG("json[%s]\n", srjson_PrintUnformatted(jdoc, jdoc->root));

+	jdoc->buf.s = srjson_PrintUnformatted(jdoc, jdoc->root);

+	if(jdoc->buf.s==NULL) {

+		LM_ERR("unable to serialize data\n");

+		return -1;

+	}

+	jdoc->buf.len = strlen(jdoc->buf.s);

+

+	LM_DBG("sending serialized data %.*s\n", jdoc->buf.len, jdoc->buf.s);

+	if (ht_dmq_send(&jdoc->buf, node)!=0) {

+		LM_ERR("unable to send data\n");

+		return -1;

+	}

+

+	LM_DBG("jdoc size[%d]\n", ht_dmq_jdoc_cell_group.size);

+

+	srjson_Delete(jdoc, jdoc_cells);

+	ht_dmq_jdoc_cell_group.jdoc_cells = srjson_CreateArray(&ht_dmq_jdoc_cell_group.jdoc);

+	if (ht_dmq_jdoc_cell_group.jdoc_cells==NULL) {

+		LM_ERR("cannot re-create json cells array! \n");

+		return -1;

+	}

+

+	ht_dmq_jdoc_cell_group.count = 0;

+	ht_dmq_jdoc_cell_group.size = dmq_cell_group_empty_size;

+

+	return 0;

+}

+

+static void ht_dmq_cell_group_destroy() {

+

+	srjson_doc_t *jdoc = &ht_dmq_jdoc_cell_group.jdoc;

+

+	if(jdoc->buf.s!=NULL) {

+		jdoc->free_fn(jdoc->buf.s);

+		jdoc->buf.s = NULL;

+	}

+	srjson_DestroyDoc(jdoc);

+

+}

+

 /**

  * @brief add notification peer

  */

@@ -59,7 +199,7 @@ int ht_dmq_initialize()

 	}

 	not_peer.callback = ht_dmq_handle_msg;

-	not_peer.init_callback = NULL;

+	not_peer.init_callback = (ht_dmq_init_sync ? ht_dmq_request_sync : NULL);

 	not_peer.description.s = "htable";

 	not_peer.description.len = 6;

 	not_peer.peer_id.s = "htable";

@@ -76,14 +216,20 @@ error:

 	return -1;

 }

-int ht_dmq_broadcast(str* body)

-{

+int ht_dmq_send(str* body, dmq_node_t* node) {

 	if (!ht_dmq_peer) {

 		LM_ERR("ht_dmq_peer is null!\n");

 		return -1;

 	}

-	LM_DBG("sending broadcast...\n");

-	ht_dmqb.bcast_message(ht_dmq_peer, body, 0, &ht_dmq_resp_callback, 1, &ht_dmq_content_type);

+	if (node) {

+		LM_DBG("sending dmq message ...\n");

+		ht_dmqb.send_message(ht_dmq_peer, body, node,

+				&ht_dmq_resp_callback, 1, &ht_dmq_content_type);

+	} else {

+		LM_DBG("sending dmq broadcast...\n");

+		ht_dmqb.bcast_message(ht_dmq_peer, body, 0,

+				&ht_dmq_resp_callback, 1, &ht_dmq_content_type);

+	}

 	return 0;

 }

@@ -138,35 +284,45 @@ int ht_dmq_handle_msg(struct sip_msg* msg, peer_reponse_t* resp, dmq_node_t* dmq

 		}

 	}

-	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, "htname")==0) {

-			htname.s = it->valuestring;

-			htname.len = strlen(htname.s);

-		} else if (strcmp(it->string, "cname")==0) {

-			cname.s = it->valuestring;

-			cname.len = strlen(cname.s);

-		} else if (strcmp(it->string, "type")==0) {

-			type = SRJSON_GET_INT(it);

-		} else if (strcmp(it->string, "strval")==0) {

-			val.s.s = it->valuestring;

-			val.s.len = strlen(val.s.s);

-		} else if (strcmp(it->string, "intval")==0) {

-			val.n = SRJSON_GET_INT(it);

-		} else if (strcmp(it->string, "mode")==0) {

-			mode = SRJSON_GET_INT(it);

+	if (unlikely(strcmp(jdoc.root->child->string, "cells")==0)) {

+		ht_dmq_handle_sync(&jdoc);

+	} else {

+

+		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, "htname")==0) {

+				htname.s = it->valuestring;

+				htname.len = strlen(htname.s);

+			} else if (strcmp(it->string, "cname")==0) {

+				cname.s = it->valuestring;

+				cname.len = strlen(cname.s);

+			} else if (strcmp(it->string, "type")==0) {

+				type = SRJSON_GET_INT(it);

+			} else if (strcmp(it->string, "strval")==0) {

+				val.s.s = it->valuestring;

+				val.s.len = strlen(val.s.s);

+			} else if (strcmp(it->string, "intval")==0) {

+				val.n = SRJSON_GET_INT(it);

+			} else if (strcmp(it->string, "mode")==0) {

+				mode = SRJSON_GET_INT(it);

+			} else {

+				LM_ERR("unrecognized field in json object\n");

+				goto invalid;

+			}

+		}

+

+		if (unlikely(action == HT_DMQ_SYNC)) {

+			ht_dmq_send_sync(dmq_node);

 		} else {

-			LM_ERR("unrecognized field in json object\n");

-			goto invalid;

+			if (ht_dmq_replay_action(action, &htname, &cname, type, &val, mode)!=0) {

+				LM_ERR("failed to replay action\n");

+				goto error;

+			}

 		}

-	}

-	if (ht_dmq_replay_action(action, &htname, &cname, type, &val, mode)!=0) {

-		LM_ERR("failed to replay action\n");

-		goto error;

 	}

 	srjson_DestroyDoc(&jdoc);

@@ -222,7 +378,7 @@ int ht_dmq_replicate_action(ht_dmq_action_t action, str* htname, str* cname, int

 	if(jdoc.buf.s!=NULL) {

 		jdoc.buf.len = strlen(jdoc.buf.s);

 		LM_DBG("sending serialized data %.*s\n", jdoc.buf.len, jdoc.buf.s);

-		if (ht_dmq_broadcast(&jdoc.buf)!=0) {

+		if (ht_dmq_send(&jdoc.buf, 0)!=0) {

 			goto error;

 		}

 		jdoc.free_fn(jdoc.buf.s);

@@ -264,9 +420,172 @@ int ht_dmq_replay_action(ht_dmq_action_t action, str* htname, str* cname, int ty

 	} else if (action==HT_DMQ_RM_CELL_RE) {

 		return ht_rm_cell_re(&val->s, ht, mode);

 	} else {

-		LM_ERR("unrecognized action");

+		LM_ERR("unrecognized action\n");

+		return -1;

+	}

+}

+

+int ht_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", HT_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 (ht_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 ht_dmq_send_sync(dmq_node_t* node) {

+	ht_t *ht;

+	ht_cell_t *it;

+	time_t now;

+	int i;

+

+	ht = ht_get_root();

+	if(ht==NULL)

+	{

+		LM_DBG("no htables to sync!\n");

+		return 0;

+	}

+

+	if (ht_dmq_cell_group_init() < 0)

 		return -1;

+

+	now = time(NULL);

+

+	while (ht != NULL)

+	{

+		if (!ht->dmqreplicate)

+			goto skip;

+

+		for(i=0; i<ht->htsize; i++)

+		{

+			ht_slot_lock(ht, i);

+			it = ht->entries[i].first;

+			while(it)

+			{

+				if(ht->htexpire > 0) {

+					if (it->expire <= now) {

+						LM_DBG("skipping expired entry\n");

+						it = it->next;

+						continue;

+					}

+				}

+

+				if (ht_dmq_cell_group_write(&ht->name, it) < 0) {

+					ht_slot_unlock(ht, i);

+					goto error;

+				}

+

+				if (ht_dmq_jdoc_cell_group.size >= dmq_cell_group_max_size) {

+					LM_DBG("sending group count[%d]size[%d]\n", ht_dmq_jdoc_cell_group.count, ht_dmq_jdoc_cell_group.size);

+					if (ht_dmq_cell_group_flush(node) < 0) {

+						ht_slot_unlock(ht, i);

+						goto error;

+					}

+				}

+

+				it = it->next;

+			}

+			ht_slot_unlock(ht, i);

+		}

+

+skip:

+		ht = ht->next;

 	}

+

+	if (ht_dmq_cell_group_flush(node) < 0)

+		goto error;

+

+	ht_dmq_cell_group_destroy();

+	return 0;

+

+error:

+	ht_dmq_cell_group_destroy();

+	return -1;

+}

+

+int ht_dmq_handle_sync(srjson_doc_t* jdoc) {

+	LM_DBG("handling sync\n");

+

+	srjson_t* cells;

+	srjson_t* cell;

+	srjson_t* it;

+	str htname;

+	str cname;

+	int type;

+	int_str val;

+	int expire;

+	ht_t* ht;

+	time_t now;

+

+

+	cells = jdoc->root->child;

+	cell = cells->child;

+

+	now = time(NULL);

+

+	while (cell) {

+		for(it=cell->child; it; it = it->next) {

+			if (strcmp(it->string, "htname")==0) {

+				htname.s = it->valuestring;

+				htname.len = strlen(htname.s);

+			} else if (strcmp(it->string, "cname")==0) {

+				cname.s = it->valuestring;

+				cname.len = strlen(cname.s);

+			} else if (strcmp(it->string, "type")==0) {

+				type = SRJSON_GET_INT(it);

+			} else if (strcmp(it->string, "strval")==0) {

+				val.s.s = it->valuestring;

+				val.s.len = strlen(val.s.s);

+			} else if (strcmp(it->string, "intval")==0) {

+				val.n = SRJSON_GET_INT(it);

+			} else if (strcmp(it->string, "expire")==0) {

+				expire = SRJSON_GET_INT(it);

+			} else {

+				LM_WARN("unrecognized field in json object\n");

+			}

+		}

+

+		ht = ht_get_table(&htname);

+		if(ht==NULL)

+			LM_WARN("unable to get table %.*s\n", ht->name.len, ht->name.s);

+

+		if (ht_set_cell_ex(ht, &cname, type, &val, 0, expire - now) < 0)

+			LM_WARN("unable to set cell %.*s in table %.*s\n", cname.len, cname.s, ht->name.len, ht->name.s);

+

+		cell = cell->next;

+	}

+	return 0;

 }

 /**

new file mode 100644

@@ -0,0 +1,280 @@

+/**

+ *

+ * Copyright (C) 2013 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 "ht_dmq.h"

+#include "ht_api.h"

+

+static str ht_dmq_content_type = str_init("application/json");

+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");

+

+typedef struct _ht_dmq_repdata {

+	int action;

+	str htname;

+	str cname;

+	int type;

+	int intval;

+	str strval;

+	int expire;

+} ht_dmq_repdata_t;

+

+dmq_api_t ht_dmqb;

+dmq_peer_t* ht_dmq_peer = NULL;

+dmq_resp_cback_t ht_dmq_resp_callback = {&ht_dmq_resp_callback_f, 0};

+

+/**

+ * @brief add notification peer

+ */

+int ht_dmq_initialize()

+{

+	dmq_peer_t not_peer;

+

+	/* load the DMQ API */

+	if (dmq_load_api(&ht_dmqb)!=0) {

+		LM_ERR("cannot load dmq api\n");

+		return -1;

+	} else {

+		LM_DBG("loaded dmq api\n");

+	}

+

+	not_peer.callback = ht_dmq_handle_msg;

+	not_peer.init_callback = NULL;

+	not_peer.description.s = "htable";

+	not_peer.description.len = 6;

+	not_peer.peer_id.s = "htable";

+	not_peer.peer_id.len = 6;

+	ht_dmq_peer = ht_dmqb.register_dmq_peer(&not_peer);

+	if(!ht_dmq_peer) {

+		LM_ERR("error in register_dmq_peer\n");

+		goto error;

+	} else {

+		LM_DBG("dmq peer registered\n");

+	}

+	return 0;

+error:

+	return -1;

+}

+

+int ht_dmq_broadcast(str* body)

+{

+	if (!ht_dmq_peer) {

+		LM_ERR("ht_dmq_peer is null!\n");

+		return -1;

+	}

+	LM_DBG("sending broadcast...\n");

+	ht_dmqb.bcast_message(ht_dmq_peer, body, 0, &ht_dmq_resp_callback, 1, &ht_dmq_content_type);

+	return 0;

+}

+

+/**

+ * @brief ht dmq callback

+ */

+int ht_dmq_handle_msg(struct sip_msg* msg, peer_reponse_t* resp, dmq_node_t* dmq_node)

+{

+	int content_length;

+	str body;

+	ht_dmq_action_t action = HT_DMQ_NONE;

+	str htname, cname;

+	int type = 0, mode = 0;

+	int_str val;

+	srjson_doc_t jdoc;

+	srjson_t *it = NULL;

+

+	/* received dmq message */

+	LM_DBG("dmq message received\n");

+

+	srjson_InitDoc(&jdoc, NULL);

+

+	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);

+

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

+		}

+	}

+

+	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, "htname")==0) {

+			htname.s = it->valuestring;

+			htname.len = strlen(htname.s);

+		} else if (strcmp(it->string, "cname")==0) {

+			cname.s = it->valuestring;

+			cname.len = strlen(cname.s);

+		} else if (strcmp(it->string, "type")==0) {

+			type = SRJSON_GET_INT(it);

+		} else if (strcmp(it->string, "strval")==0) {

+			val.s.s = it->valuestring;

+			val.s.len = strlen(val.s.s);

+		} else if (strcmp(it->string, "intval")==0) {

+			val.n = SRJSON_GET_INT(it);

+		} else if (strcmp(it->string, "mode")==0) {

+			mode = SRJSON_GET_INT(it);

+		} else {

+			LM_ERR("unrecognized field in json object\n");

+			goto invalid;

+		}

+	}

+

+	if (ht_dmq_replay_action(action, &htname, &cname, type, &val, mode)!=0) {

+		LM_ERR("failed to replay action\n");

+		goto error;

+	}

+

+	srjson_DestroyDoc(&jdoc);

+	resp->reason = dmq_200_rpl;

+	resp->resp_code = 200;

+	return 0;

+

+invalid:

+	srjson_DestroyDoc(&jdoc);

+	resp->reason = dmq_400_rpl;

+	resp->resp_code = 400;

+	return 0;

+

+error:

+	srjson_DestroyDoc(&jdoc);

+	resp->reason = dmq_500_rpl;

+	resp->resp_code = 500;

+	return 0;

+}