Merge branch 'feature/media_exchange'

Author: razvancrainea

modules/b2b_entities/b2b_entities.c

@@ -35,6 +35,7 @@
 #include "../../dprint.h"
 #include "../../error.h"
 #include "../../ut.h"
+#include "../../pt.h"
 #include "../../script_cb.h"
 #include "../../parser/parse_from.h"
 #include "../dialog/dlg_load.h"
@@ -411,11 +412,12 @@ int b2b_restore_logic_info(enum b2b_entity_type type, str* key,
 }
 
 int b2b_update_b2bl_param(enum b2b_entity_type type, str* key,
-		str* param)
+		str* param, int replicate)
 {
 	b2b_dlg_t* dlg;
 	b2b_table table;
 	unsigned int hash_index, local_index;
+	int unlock = 1;
 
 	if(!param)
 	{
@@ -442,19 +444,25 @@ int b2b_update_b2bl_param(enum b2b_entity_type type, str* key,
 		LM_ERR("Wrong format for b2b key [%.*s]\n", key->len, key->s);
 		return -1;
 	}
-	lock_get(&table[hash_index].lock);
+	if (table[hash_index].locked_by != process_no)
+		lock_get(&table[hash_index].lock);
+	else
+		unlock = 0;
+
 	dlg = b2b_search_htable(table, hash_index, local_index);
 	if(dlg == NULL)
 	{
 		LM_ERR("No dialog found\n");
-		lock_release(&table[hash_index].lock);
+		if (unlock)
+			lock_release(&table[hash_index].lock);
 		return -1;
 	}
 	memcpy(dlg->param.s, param->s, param->len);
 	dlg->param.len = param->len;
-	lock_release(&table[hash_index].lock);
+	if (unlock)
+		lock_release(&table[hash_index].lock);
 
-	if (b2be_cluster)
+	if (b2be_cluster && replicate)
 		replicate_entity_update(dlg, type, hash_index, param, NULL);
 
 	return 0;

modules/b2b_entities/b2be_clustering.c

@@ -37,6 +37,7 @@
 #include "b2be_clustering.h"
 #include "b2b_entities.h"
 #include "b2be_db.h"
+#include "../../pt.h"
 
 #define NO_REPL_CONSTANT_STRS 10
 
@@ -456,14 +457,16 @@ int receive_entity_create(bin_packet_t *packet, b2b_dlg_t *dlg, int type,
 
 	lock_get(&htable[h_idx].lock);
 
-	b2b_run_cb(new_dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_CREATE, packet);
-
 	new_key = b2b_htable_insert(htable, new_dlg, hash_index, type, 1, 1);
 	if (new_key == NULL) {
 		LM_ERR("Failed to insert new record\n");
 		goto error;
 	}
 
+	htable[h_idx].locked_by = process_no;
+	b2b_run_cb(new_dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_CREATE, packet);
+	htable[h_idx].locked_by = -1;
+
 	lock_release(&htable[h_idx].lock);
 
 	pkg_free(new_key);
@@ -551,7 +554,9 @@ int receive_entity_update(bin_packet_t *packet)
 	if (packet->type == REPL_ENTITY_UPDATE) {
 		unpack_update_fields(packet, dlg);
 
+		htable[hash_index].locked_by = process_no;
 		b2b_run_cb(dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_UPDATE, packet);
+		htable[hash_index].locked_by = -1;
 	} else {
 		rc = recv_b2bl_param_update(packet, dlg);
 	}
@@ -605,7 +610,9 @@ int receive_entity_delete(bin_packet_t *packet)
 		return 0;
 	}
 
+	htable[hash_index].locked_by = process_no;
 	b2b_run_cb(dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_DELETE, packet);
+	htable[hash_index].locked_by = -1;
 
 	b2b_entity_db_delete(type, dlg);
 	b2b_delete_record(dlg, htable, hash_index);

modules/b2b_entities/b2be_load.h

@@ -124,7 +124,7 @@ typedef int (*b2b_restore_linfo_t)(enum b2b_entity_type type, str* key,
 typedef int (*b2b_reg_cb_t) (b2b_cb_t cb, int cb_type, str *mod_name);
 
 typedef int (*b2b_update_b2bl_param_t)(enum b2b_entity_type type, str* key,
-		str* param);
+		str* param, int replicate);
 typedef int (*b2b_get_b2bl_key_t)(str* callid, str* from_tag, str* to_tag,
 		str* entity_key, str* tuple_key);
 

modules/b2b_entities/dlg.h

@@ -117,6 +117,7 @@ typedef struct b2b_entry
 {
 	b2b_dlg_t* first;
 	gen_lock_t lock;
+	int locked_by;
 	int checked;
 }b2b_entry_t;
 

modules/b2b_entities/doc/b2b_entities_devel.xml

@@ -218,7 +218,7 @@ typedef int (*b2b_restore_linfo_t)(enum b2b_entity_type type, str* key,
 			<programlisting format="linespecific">
 ...
 typedef int (*b2b_update_b2bl_param_t)(enum b2b_entity_type type, str* key,
-		str* param);
+		str* param, int replicate);
 ...
 </programlisting>
 		<para>

modules/b2b_logic/logic.c

@@ -4090,7 +4090,7 @@ int b2bl_bridge_2calls(str* key1, str* key2)
 	e1->no = 0;
 	e2->no = 1;
 
-	if(b2b_api.update_b2bl_param(e2->type, &e2->key, tuple->key) < 0)
+	if(b2b_api.update_b2bl_param(e2->type, &e2->key, tuple->key, 1) < 0)
 	{
 		LM_ERR("Failed to update b2bl parameter in b2b_entities\n");
 		goto error;

modules/dialog/dialog.c

@@ -594,6 +594,10 @@ int load_dlg( struct dlg_binds *dlgb )
 	dlgb->dlg_ref = _ref_dlg;
 	dlgb->dlg_unref = unref_dlg_destroy_safe;
 
+	dlgb->get_direction = get_dlg_direction;
+	dlgb->get_dlg_by_callid = get_dlg_by_callid;
+	dlgb->send_indialog_request = send_indialog_request;
+
 	dlgb->get_rr_param = get_rr_param;
 
 	/* dlg context functions */
@@ -1563,20 +1567,26 @@ int pv_get_dlg_timeout(struct sip_msg *msg, pv_param_t *param,
 int pv_get_dlg_dir(struct sip_msg *msg, pv_param_t *param,
 		pv_value_t *res)
 {
-	struct dlg_cell *dlg;
+	int dir;
 
 	if(res==NULL)
 		return -1;
 
-	if ( (dlg=get_current_dialog())==NULL || ctx_lastdstleg_get()<0)
-		return pv_get_null( msg, param, res);
-
-	if (ctx_lastdstleg_get()==0) {
-		res->rs.s = "upstream";
-		res->rs.len = 8;
-	} else {
-		res->rs.s = "downstream";
-		res->rs.len = 10;
+	dir = get_dlg_direction();
+	switch (dir) {
+		case DLG_DIR_NONE:
+			return pv_get_null( msg, param, res);
+		case DLG_DIR_UPSTREAM:
+			res->rs.s = "upstream";
+			res->rs.len = 8;
+			break;
+		case DLG_DIR_DOWNSTREAM:
+			res->rs.s = "downstream";
+			res->rs.len = 10;
+			break;
+		default:
+			LM_BUG("unknwn dlg direction %d!\n", dir);
+			return -1;
 	}
 
 	res->flags = PV_VAL_STR;

modules/dialog/dlg_handlers.h

@@ -127,6 +127,12 @@ int terminate_dlg(str *callid, unsigned int h_entry, unsigned int h_id,
 typedef int (*terminate_dlg_f)(str *callid, unsigned int h_entry, unsigned int h_id,
 	str *reason);
 
+typedef int (*indialog_reply_f) (struct sip_msg *msg, int statuscode, void *param);
+typedef int (*send_indialog_req_f)(struct dlg_cell *dlg, str *method,
+		int leg, str *body, str *ct, indialog_reply_f func, void *param);
+int send_indialog_request(struct dlg_cell *dlg, str *method,
+		int leg, str *body, str *ct, indialog_reply_f func, void *param);
+
 void unreference_dialog(void *dialog);
 
 static inline int parse_dlg_rr_param(char *p, char *end,

modules/dialog/dlg_hash.h

@@ -206,6 +206,10 @@ extern int dlg_enable_stats;
 #define other_leg(dlg, l) \
 	(l == DLG_CALLER_LEG? callee_idx(dlg): DLG_CALLER_LEG)
 
+#define dlg_get_out_sdp(_dlg, _dst) \
+	((_dlg)->legs[(_dst)].out_sdp.s? \
+	 (_dlg)->legs[(_dst)].out_sdp:(_dlg)->legs[other_leg(_dlg, _dst)].in_sdp)
+
 
 #define ctx_dialog_get() \
 	((struct dlg_cell*)context_get_ptr(CONTEXT_GLOBAL,current_processing_ctx,ctx_dlg_idx) )
@@ -403,6 +407,8 @@ struct dlg_cell* get_dlg_by_val(str *attr, str *val);
 
 struct dlg_cell* get_dlg_by_callid( str *callid, int active_only);
 
+int get_dlg_direction(void);
+
 void link_dlg(struct dlg_cell *dlg, int extra_refs);
 
 #define _link_dlg_unsafe(d_entry, dlg) \

modules/dialog/dlg_load.h

@@ -35,7 +35,9 @@
 #include "../../sr_module.h"
 
 typedef struct dlg_cell *(*get_dlg_f) (void);
+typedef struct dlg_cell *(*get_dlg_by_callid_f) (str *, int);
 typedef int (*match_dialog_f) (struct sip_msg *msg, int _seq_match_mode);
+typedef int (*get_direction_f) (void);
 
 struct dlg_binds {
 	register_dlgcb_f     register_dlgcb;
@@ -60,6 +62,10 @@ struct dlg_binds {
 	ref_dlg_f            dlg_ref;
 	unref_dlg_f          dlg_unref;
 
+	get_direction_f      get_direction;
+	get_dlg_by_callid_f  get_dlg_by_callid;
+	send_indialog_req_f  send_indialog_request;
+
 	get_rr_param_f       get_rr_param;
 
 	dlg_ctx_register_int_f dlg_ctx_register_int;