dialog: replicate dialog values

modules/dialog/dlg_cb.h

@@ -241,6 +241,10 @@ typedef int (*register_dlgcb_f)(struct dlg_cell* dlg, int cb_types,
  * Registration:  per-dialog, "dlg" must be given
  * Trigger count: 0 - N times per dialog (e.g. for every update and create
  *                replicated packet received on the network)
+ * Params:
+ *   - (str *)params->dlg_data will hold the name of the variable that is being
+ *     replicated, if it was replicated after the dialog was confirmed, or
+ *     NULL if the variables were replicated in bulk
  */
 #define DLGCB_PROCESS_VARS (1<<14)
 

modules/dialog/dlg_replication.c

@@ -730,6 +730,53 @@ int dlg_replicated_cseq_updated(bin_packet_t *packet)
 	LM_ERR("malformed cseq update packet for %.*s\n", call_id.len, call_id.s);
 	return -1;
 }
+
+int dlg_replicated_value(bin_packet_t *packet)
+{
+	str call_id, name;
+	struct dlg_cell *dlg;
+	unsigned int h_id;
+	int h_entry, type;
+	struct dlg_entry *d_entry;
+	int_str val;
+
+	DLG_BIN_POP(str, packet, call_id, malformed);
+	DLG_BIN_POP(int, packet, h_id, malformed);
+	DLG_BIN_POP(str, packet, name, malformed);
+	DLG_BIN_POP(int, packet, type, malformed);
+	if (type == DLG_VAL_TYPE_STR)
+		DLG_BIN_POP(str, packet, val.s, malformed);
+	else
+		DLG_BIN_POP(int, packet, val.n, malformed);
+
+	LM_DBG("Updating cseq for dialog with callid: %.*s\n", call_id.len, call_id.s);
+	h_entry = dlg_hash(&call_id);
+	d_entry = &(d_table->entries[h_entry]);
+
+	dlg_lock(d_table, d_entry);
+
+	dlg = lookup_dlg_unsafe(h_entry, h_id);
+	if (!dlg) {
+		LM_DBG("unable to find dialog %.*s [%u:%d]\n",
+				call_id.len, call_id.s, h_id, h_entry);
+		dlg_unlock(d_table, d_entry);
+		return -1;
+	}
+	lock_start_write(dlg->vals_lock);
+	if (store_dlg_value_unsafe(dlg, &name, &val, type) < 0)
+		LM_ERR("cannot store dlg value\n");
+	else
+		run_dlg_callbacks(DLGCB_PROCESS_VARS, dlg,
+				NULL, DLG_DIR_NONE, -1, &name, 1, 0);
+	lock_stop_write(dlg->vals_lock);
+
+	dlg_unlock(d_table, d_entry);
+
+	return 0;
+malformed:
+	LM_ERR("malformed cseq update packet for %.*s\n", call_id.len, call_id.s);
+	return -1;
+}
 #undef DLG_BIN_POP
 
 
@@ -994,6 +1041,33 @@ void replicate_dialog_cseq_updated(struct dlg_cell *dlg, int leg)
 	LM_ERR("Failed to replicate dialog cseq update\n");
 }
 
+void replicate_dialog_value(struct dlg_cell *dlg, str *name, int_str *val, int type)
+{
+	bin_packet_t packet;
+
+	if (bin_init(&packet, &dlg_repl_cap, REPLICATION_DLG_VALUE,
+			BIN_VERSION, 512) != 0)
+		goto error;
+
+	bin_push_str(&packet, &dlg->callid);
+	bin_push_int(&packet, dlg->h_id);
+	bin_push_str(&packet, name);
+	bin_push_int(&packet, type);
+	if (type == DLG_VAL_TYPE_STR)
+		bin_push_str(&packet, &val->s);
+	else
+		bin_push_int(&packet, val->n);
+
+	DLG_CLUSTER_SEND(packet, dialog_repl_cluster, error_free);
+
+	bin_free_packet(&packet);
+	return;
+error_free:
+	bin_free_packet(&packet);
+error:
+	LM_ERR("Failed to replicate dialog values\n");
+}
+
 #undef DLG_CLUSTER_SEND
 
 void receive_dlg_repl(bin_packet_t *pkt)
@@ -1030,6 +1104,11 @@ void receive_dlg_repl(bin_packet_t *pkt)
 
 		rc = dlg_replicated_cseq_updated(pkt);
 		break;
+	case REPLICATION_DLG_VALUE:
+		ensure_bin_version(pkt, BIN_VERSION);
+
+		rc = dlg_replicated_value(pkt);
+		break;
 	case SYNC_PACKET_TYPE:
 		if (ver != DLG_BIN_V3)
 			ensure_bin_version(pkt, BIN_VERSION);

modules/dialog/dlg_replication.h

@@ -32,6 +32,7 @@
 #define REPLICATION_DLG_UPDATED		2
 #define REPLICATION_DLG_DELETED		3
 #define REPLICATION_DLG_CSEQ		4
+#define REPLICATION_DLG_VALUE		5
 
 #define DLG_BIN_V3      3
 #define DLG_BIN_V4      4
@@ -56,6 +57,7 @@ void replicate_dialog_created(struct dlg_cell *dlg);
 void replicate_dialog_updated(struct dlg_cell *dlg);
 void replicate_dialog_deleted(struct dlg_cell *dlg);
 void replicate_dialog_cseq_updated(struct dlg_cell *dlg, int leg);
+void replicate_dialog_value(struct dlg_cell *dlg, str *name, int_str *val, int type);
 
 int dlg_replicated_create(bin_packet_t *packet, struct dlg_cell *cell,
 	str *ftag, str *ttag, unsigned int hid, int safe, int from_sync);

modules/dialog/dlg_vals.c

@@ -23,6 +23,7 @@
 #include "../../pt.h"
 #include "dlg_vals.h"
 #include "dlg_hash.h"
+#include "dlg_replication.h"
 
 
 
@@ -134,6 +135,9 @@ int store_dlg_value(struct dlg_cell *dlg, str *name, int_str *val, int type)
 	ret = store_dlg_value_unsafe(dlg,name,val,type);
 	lock_stop_write(dlg->vals_lock);
 
+	if (ret == 0 && dlg->state >= DLG_STATE_CONFIRMED)
+		replicate_dialog_value(dlg, name, val, type);
+
 	return ret;
 }