Rtpengine dmq sync

src/modules/rtpengine/rtpengine_dmq.c

@@ -22,6 +22,7 @@
 
 #include "rtpengine.h"
 #include "rtpengine_dmq.h"
+#include "../../core/timer.h"
 
 static str rtpengine_dmq_content_type = str_init("application/json");
 static str rtpengine_dmq_200_rpl = str_init("OK");
@@ -31,7 +32,8 @@ static str rtpengine_dmq_500_rpl = str_init("Server Internal Error");
 dmq_api_t rtpengine_dmqb;
 dmq_peer_t *rtpengine_dmq_peer = NULL;
 
-int rtpengine_dmq_request_sync(dmq_node_t *node);
+int rtpengine_dmq_request_sync();
+int rtpengine_dmq_send(str *body, dmq_node_t *node);
 
 /**
 * @brief add rtpengine notification peer
@@ -64,9 +66,44 @@ int rtpengine_dmq_init()
 	return 0;
 }
 
-int rtpengine_dmq_request_sync(dmq_node_t *node)
+int rtpengine_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", RTPENGINE_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(rtpengine_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 rtpengine_dmq_handle_msg(
@@ -197,6 +234,10 @@ int rtpengine_dmq_handle_msg(
 				goto error;
 			break;
 		case RTPENGINE_DMQ_SYNC:
+			if(rtpengine_dmq_replicate_sync(node) != 0) {
+				goto error;
+			}
+			break;
 		case RTPENGINE_DMQ_NONE:
 			break;
 	}
@@ -276,7 +317,7 @@ int rtpengine_dmq_replicate_action(rtpengine_dmq_action_t action, str callid,
 	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(rtpengine_dmq_send(&jdoc.buf, 0) != 0) {
+		if(rtpengine_dmq_send(&jdoc.buf, node) != 0) {
 			goto error;
 		}
 		jdoc.free_fn(jdoc.buf.s);
@@ -310,3 +351,51 @@ int rtpengine_dmq_replicate_remove(str callid, str viabranch)
 	return rtpengine_dmq_replicate_action(
 			RTPENGINE_DMQ_REMOVE, callid, viabranch, NULL, NULL);
 }
+
+int rtpengine_dmq_replicate_sync(dmq_node_t *node)
+{
+	int i;
+	struct rtpengine_hash_entry *entry;
+	struct rtpengine_hash_table *hash_table;
+
+	LM_DBG("replicating all hash entries to dmq peers\n");
+
+	if(!rtpengine_hash_table_sanity_checks()) {
+		LM_ERR("sanity checks failed\n");
+		return -1;
+	}
+	hash_table = rtpengine_hash_table_get();
+	if(!hash_table) {
+		LM_ERR("NULL rtpengine_hash_table\n");
+		return -1;
+	}
+
+	for(i = 0; i < hash_table->size; i++) {
+		lock_get(&hash_table->row_locks[i]);
+
+		entry = hash_table->row_entry_list[i];
+
+		while(entry) {
+			if(entry->tout >= get_ticks()) {
+				LM_DBG("replicating hash entry callid=%.*s viabranch=%.*s\n",
+						entry->callid.len, entry->callid.s,
+						entry->viabranch.len, entry->viabranch.s);
+
+				if(rtpengine_dmq_replicate_action(RTPENGINE_DMQ_INSERT,
+						   entry->callid, entry->viabranch, entry, node)
+						!= 0) {
+					LM_ERR("failed to replicate hash entry callid=%.*s "
+						   "viabranch=%.*s\n",
+							entry->callid.len, entry->callid.s,
+							entry->viabranch.len, entry->viabranch.s);
+				}
+			}
+
+			entry = entry->next;
+		}
+
+		lock_release(&hash_table->row_locks[i]);
+	}
+
+	return 0;
+}

src/modules/rtpengine/rtpengine_dmq.h

@@ -48,5 +48,5 @@ int rtpengine_dmq_replicate_action(rtpengine_dmq_action_t action, str callid,
 int rtpengine_dmq_replicate_insert(
 		str callid, str viabranch, struct rtpengine_hash_entry *entry);
 int rtpengine_dmq_replicate_remove(str callid, str viabranch);
-int rtpengine_dmq_replicate_sync();
+int rtpengine_dmq_replicate_sync(dmq_node_t *node);
 #endif

src/modules/rtpengine/rtpengine_hash.c

@@ -7,7 +7,7 @@
 #include "../../core/timer.h"
 
 
-static struct rtpengine_hash_table *rtpengine_hash_table;
+struct rtpengine_hash_table *rtpengine_hash_table;
 
 /* from sipwise rtpengine */
 static unsigned int str_hash(str s)
@@ -156,6 +156,11 @@ int rtpengine_hash_table_destroy()
 	return 1;
 }
 
+struct rtpengine_hash_table *rtpengine_hash_table_get()
+{
+	return rtpengine_hash_table;
+}
+
 int rtpengine_hash_table_insert(
 		str callid, str viabranch, struct rtpengine_hash_entry *value)
 {

src/modules/rtpengine/rtpengine_hash.h

@@ -28,8 +28,11 @@ struct rtpengine_hash_table
 };
 
 
+extern struct rtpengine_hash_table *rtpengine_hash_table;
+
 int rtpengine_hash_table_init(int size);
 int rtpengine_hash_table_destroy();
+struct rtpengine_hash_table *rtpengine_hash_table_get();
 int rtpengine_hash_table_insert(
 		str callid, str viabranch, struct rtpengine_hash_entry *value);
 int rtpengine_hash_table_remove(str callid, str viabranch, enum rtpe_operation);