Adding dialog replication via DMQ

Updating code base to actual state.
Adding repliation for DLG_STATE_EARLY to prevent hanging up unconfirmed dialogs and its infinite replication.
Fixing dialog removal on peers to prevent die on timeout and possible infinite replication.
Fixing dialog removal on peers when master node cleanly restart.

modules/dialog/dialog.c

@@ -79,6 +79,7 @@
 #include "dlg_var.h"
 #include "dlg_transfer.h"
 #include "dlg_cseq.h"
+#include "dlg_dmq.h"
 
 MODULE_VERSION
 
@@ -107,6 +108,10 @@ static int dlg_timer_procs = 0;
 static int _dlg_track_cseq_updates = 0;
 int dlg_ka_failed_limit = 1;
 
+int dlg_enable_dmq = 0;
+int dlg_id_offset = 1;
+int dlg_id_increment = 1;
+
 int dlg_event_rt[DLG_EVENTRT_MAX];
 
 str dlg_bridge_controller = str_init("sip:controller@kamailio.org");
@@ -296,6 +301,10 @@ static param_export_t mod_params[]={
 	{ "lreq_callee_headers",   PARAM_STR, &dlg_lreq_callee_headers  },
 	{ "db_skip_load",          INT_PARAM, &db_skip_load             },
 	{ "ka_failed_limit",       INT_PARAM, &dlg_ka_failed_limit      },
+	{ "enable_dmq",            INT_PARAM, &dlg_enable_dmq           },
+	{ "id_offset",             INT_PARAM, &dlg_id_offset            },
+	{ "id_increment",          INT_PARAM, &dlg_id_increment         },
+
 	{ 0,0,0 }
 };
 
@@ -712,6 +721,11 @@ static int mod_init(void)
 		dlg_register_cseq_callbacks();
 	}
 
+	if (dlg_enable_dmq>0 && dlg_dmq_initialize()!=0) {
+		LM_ERR("failed to initialize dmq integration\n");
+		return -1;
+	}
+
 	return 0;
 }
 

modules/dialog/dlg_db_handler.c

@@ -426,6 +426,12 @@ static int load_dialog_info_from_db(int dlg_hash_size, int fetch_num_rows)
 				srjson_DestroyDoc(&jdoc);
 			}
 			dlg->iflags = (unsigned int)VAL_INT(values+22);
+
+			if (!dlg->bind_addr[DLG_CALLER_LEG] || !dlg->bind_addr[DLG_CALLEE_LEG]) {
+				/* non-local socket, probably not our dialog */
+				dlg->iflags &= ~DLG_IFLAG_DMQ_SYNC;
+			}
+
 			if(dlg->state==DLG_STATE_DELETED) {
 				/* end_ts used for force clean up not stored - set it to now */
 				dlg->end_ts = (unsigned int)time(0);